[teiid-commits] teiid SVN: r3502 - in trunk: client/src/main/java/org/teiid/client and 20 other directories.

Author: shawkins Date: 2011-09-21 00:13:59 -0400 (Wed, 21 Sep 2011) New Revision: 3502 Added: trunk/engine/src/main/java/org/teiid/common/buffer/AutoCleanupUtil.java trunk/engine/src/main/java/org/teiid/common/buffer/Cache.java trunk/engine/src/main/java/org/teiid/common/buffer/CacheEntry.java trunk/engine/src/main/java/org/teiid/common/buffer/LightWeightCopyOnWriteList.java trunk/engine/src/main/java/org/teiid/common/buffer/Serializer.java trunk/engine/src/main/java/org/teiid/common/buffer/impl/BitSetTree.java trunk/engine/src/main/java/org/teiid/common/buffer/impl/FileStoreCache.java trunk/engine/src/main/java/org/teiid/dqp/internal/process/SerializableTupleBatch.java trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestBitSetTree.java trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestFileStoreCache.java Removed: trunk/engine/src/test/java/org/teiid/core/ Modified: trunk/cache-jbosscache/src/main/java/org/teiid/cache/jboss/TupleBatchCacheLoader.java trunk/client/src/main/java/org/teiid/client/BatchSerializer.java trunk/client/src/main/java/org/teiid/client/ResultsMessage.java trunk/client/src/main/java/org/teiid/jdbc/ResultSetImpl.java trunk/client/src/test/java/org/teiid/client/TestBatchSerializer.java trunk/client/src/test/java/org/teiid/jdbc/TestAllResultsImpl.java trunk/client/src/test/java/org/teiid/jdbc/TestCallableStatement.java trunk/common-core/src/main/java/org/teiid/core/types/DataTypeManager.java trunk/engine/src/main/java/org/teiid/common/buffer/BatchManager.java trunk/engine/src/main/java/org/teiid/common/buffer/FileStore.java trunk/engine/src/main/java/org/teiid/common/buffer/FileStoreInputStreamFactory.java trunk/engine/src/main/java/org/teiid/common/buffer/LobManager.java trunk/engine/src/main/java/org/teiid/common/buffer/SPage.java trunk/engine/src/main/java/org/teiid/common/buffer/STree.java trunk/engine/src/main/java/org/teiid/common/buffer/TupleBatch.java trunk/engine/src/main/java/org/teiid/common/buffer/TupleBrowser.java trunk/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java trunk/engine/src/main/java/org/teiid/common/buffer/impl/FileStorageManager.java trunk/engine/src/main/java/org/teiid/common/buffer/impl/MemoryStorageManager.java trunk/engine/src/main/java/org/teiid/common/buffer/impl/SizeUtility.java trunk/engine/src/main/java/org/teiid/common/buffer/impl/SplittableStorageManager.java trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorWorkItem.java trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java trunk/engine/src/main/java/org/teiid/dqp/message/AtomicResultsMessage.java trunk/engine/src/main/java/org/teiid/query/processor/relational/EnhancedSortMergeJoinStrategy.java trunk/engine/src/main/java/org/teiid/query/tempdata/TempTable.java trunk/engine/src/test/java/org/teiid/common/buffer/BufferManagerFactory.java trunk/engine/src/test/java/org/teiid/common/buffer/TestLobManager.java trunk/engine/src/test/java/org/teiid/common/buffer/TestSTree.java trunk/engine/src/test/java/org/teiid/common/buffer/TestTupleBuffer.java trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestFileStorageManager.java trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestSizeUtility.java trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestSplittableStorageManager.java trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestCachedResults.java trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java trunk/engine/src/test/java/org/teiid/dqp/service/AutoGenDataService.java trunk/engine/src/test/java/org/teiid/query/processor/TestTempTables.java trunk/engine/src/test/java/org/teiid/query/processor/TestVirtualDepJoin.java trunk/engine/src/test/java/org/teiid/query/sql/symbol/TestElementSymbol.java trunk/runtime/src/main/java/org/teiid/services/BufferServiceImpl.java trunk/runtime/src/test/java/org/teiid/dqp/service/buffer/TestLocalBufferService.java Log: TEIID-1750 fixed update leaking with transactional temp tables. switching the buffermanager caching logic to focus solely on memory management. minimized the memory footprint of the buffermanager and managed batches. switching the disk cache logic to a block approach - which will also allow for the introduction of off-heap caching. adding a batched mode for ordered temp table inserts. Modified: trunk/cache-jbosscache/src/main/java/org/teiid/cache/jboss/TupleBatchCacheLoader.java =================================================================== --- trunk/cache-jbosscache/src/main/java/org/teiid/cache/jboss/TupleBatchCacheLoader.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/cache-jbosscache/src/main/java/org/teiid/cache/jboss/TupleBatchCacheLoader.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -31,6 +31,7 @@ import org.teiid.common.buffer.TupleBatch; import org.teiid.common.buffer.TupleBuffer; import org.teiid.core.TeiidRuntimeException; +import org.teiid.dqp.internal.process.SerializableTupleBatch; public class TupleBatchCacheLoader extends ClusteredTupleBatchCacheLoader { @@ -58,8 +59,7 @@ if (tb != null) { Map map = new HashMap(); TupleBatch b = tb.getBatch(row); - b.preserveTypes(); - map.put(id, b); + map.put(id, new SerializableTupleBatch(b, tb.getTypes())); return map; } } Modified: trunk/client/src/main/java/org/teiid/client/BatchSerializer.java =================================================================== --- trunk/client/src/main/java/org/teiid/client/BatchSerializer.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/client/src/main/java/org/teiid/client/BatchSerializer.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -29,6 +29,7 @@ import java.math.BigInteger; import java.sql.Time; import java.sql.Timestamp; +import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.List; @@ -75,23 +76,23 @@ * @throws IOException * @since 4.2 */ - static void writeIsNullData(ObjectOutput out, int col, List[] batch) throws IOException { - int numBytes = batch.length / 8, row = 0, currentByte = 0; + static void writeIsNullData(ObjectOutput out, int col, List<? extends List<?>> batch) throws IOException { + int numBytes = batch.size() / 8, row = 0, currentByte = 0; for (int byteNum = 0; byteNum < numBytes; byteNum++, row+=8) { - currentByte = (batch[row].get(col) == null) ? 0x80 : 0; - if (batch[row+1].get(col) == null) currentByte |= 0x40; - if (batch[row+2].get(col) == null) currentByte |= 0x20; - if (batch[row+3].get(col) == null) currentByte |= 0x10; - if (batch[row+4].get(col) == null) currentByte |= 0x08; - if (batch[row+5].get(col) == null) currentByte |= 0x04; - if (batch[row+6].get(col) == null) currentByte |= 0x02; - if (batch[row+7].get(col) == null) currentByte |= 0x01; + currentByte = (batch.get(row).get(col) == null) ? 0x80 : 0; + if (batch.get(row+1).get(col) == null) currentByte |= 0x40; + if (batch.get(row+2).get(col) == null) currentByte |= 0x20; + if (batch.get(row+3).get(col) == null) currentByte |= 0x10; + if (batch.get(row+4).get(col) == null) currentByte |= 0x08; + if (batch.get(row+5).get(col) == null) currentByte |= 0x04; + if (batch.get(row+6).get(col) == null) currentByte |= 0x02; + if (batch.get(row+7).get(col) == null) currentByte |= 0x01; out.write(currentByte); } - if (batch.length % 8 > 0) { + if (batch.size() % 8 > 0) { currentByte = 0; - for (int mask = 0x80; row < batch.length; row++, mask >>= 1) { - if (batch[row].get(col) == null) currentByte |= mask; + for (int mask = 0x80; row < batch.size(); row++, mask >>= 1) { + if (batch.get(row).get(col) == null) currentByte |= mask; } out.write(currentByte); } @@ -126,22 +127,22 @@ * @since 4.2 */ private static class ColumnSerializer { - public void writeColumn(ObjectOutput out, int col, List[] batch) throws IOException { + public void writeColumn(ObjectOutput out, int col, List<? extends List<?>> batch) throws IOException { writeIsNullData(out, col, batch); Object obj = null; - for (int i = 0; i < batch.length; i++) { - obj = batch[i].get(col); + for (int i = 0; i < batch.size(); i++) { + obj = batch.get(i).get(col); if (obj != null) { writeObject(out, obj); } } } - public void readColumn(ObjectInput in, int col, List[] batch, byte[] isNull) throws IOException, ClassNotFoundException { + public void readColumn(ObjectInput in, int col, List<List<Object>> batch, byte[] isNull) throws IOException, ClassNotFoundException { readIsNullData(in, isNull); - for (int i = 0; i < batch.length; i++) { + for (int i = 0; i < batch.size(); i++) { if (!isNullObject(isNull, i)) { - batch[i].set(col, DataTypeManager.getCanonicalValue(readObject(in))); + batch.get(i).set(col, DataTypeManager.getCanonicalValue(readObject(in))); } } } @@ -203,13 +204,14 @@ /* This implementation compacts the isNull and boolean data for non-null values into a byte[] * by using a 8 bit mask that is bit-shifted to mask each value. */ - public void writeColumn(ObjectOutput out, int col, List[] batch) throws IOException { + @Override + public void writeColumn(ObjectOutput out, int col, List<? extends List<?>> batch) throws IOException { int currentByte = 0; int mask = 0x80; Object obj; - for (int row = 0; row < batch.length; row++) { + for (int row = 0; row < batch.size(); row++) { // Write the isNull value - obj = batch[row].get(col); + obj = batch.get(row).get(col); if (obj == null ) { currentByte |= mask; } @@ -241,10 +243,13 @@ } } - public void readColumn(ObjectInput in, int col, List[] batch, byte[] isNull) throws IOException, ClassNotFoundException { + @Override + public void readColumn(ObjectInput in, int col, + List<List<Object>> batch, byte[] isNull) throws IOException, + ClassNotFoundException { int currentByte = 0, mask = 0; // Initialize the mask so that it is reset in the loop boolean isNullVal; - for (int row = 0; row < batch.length; row++) { + for (int row = 0; row < batch.size(); row++) { if (mask == 0) { // If we used up the byte, read the next one, and reset the mask currentByte = in.read(); @@ -257,7 +262,7 @@ currentByte = in.read(); mask = 0x80; } - batch[row].set(col, ((currentByte & mask) == 0) ? Boolean.FALSE : Boolean.TRUE); + batch.get(row).set(col, ((currentByte & mask) == 0) ? Boolean.FALSE : Boolean.TRUE); mask >>= 1; } } @@ -354,17 +359,12 @@ return cs; } - public static void writeBatch(ObjectOutput out, String[] types, List[] batch) throws IOException { - // If there are no type hints, simply use the default mechanism to serialize - if (types == null || types.length == 0) { - out.writeObject(batch); - return; - } + public static void writeBatch(ObjectOutput out, String[] types, List<? extends List<?>> batch) throws IOException { if (batch == null) { out.writeInt(-1); } else { - out.writeInt(batch.length); - if (batch.length > 0) { + out.writeInt(batch.size()); + if (batch.size() > 0) { int columns = types.length; out.writeInt(columns); for(int i = 0; i < columns; i++) { @@ -374,8 +374,8 @@ } catch (ClassCastException e) { Object obj = null; String objectClass = null; - objectSearch: for (int row = 0; row < batch.length; row++) { - obj = batch[row].get(i); + objectSearch: for (int row = 0; row < batch.size(); row++) { + obj = batch.get(row).get(i); if (obj != null) { objectClass = obj.getClass().getName(); break objectSearch; @@ -388,21 +388,17 @@ } } - public static List[] readBatch(ObjectInput in, String[] types) throws IOException, ClassNotFoundException { - // If there are no type hints, use the default mechanism to deserialize - if (types == null || types.length == 0) { - return (List[])in.readObject(); - } + public static List<List<Object>> readBatch(ObjectInput in, String[] types) throws IOException, ClassNotFoundException { int rows = in.readInt(); if (rows == 0) { - return new List[0]; + return new ArrayList<List<Object>>(0); } else if (rows > 0) { int columns = in.readInt(); - List[] batch = new List[rows]; + List<List<Object>> batch = new ArrayList<List<Object>>(rows); int numBytes = rows/8; int extraRows = rows % 8; for (int currentRow = 0; currentRow < rows; currentRow++) { - batch[currentRow] = Arrays.asList(new Object[columns]); + batch.add(currentRow, Arrays.asList(new Object[columns])); } byte[] isNullBuffer = new byte[(extraRows > 0) ? numBytes + 1: numBytes]; for (int col = 0; col < columns; col++) { Modified: trunk/client/src/main/java/org/teiid/client/ResultsMessage.java =================================================================== --- trunk/client/src/main/java/org/teiid/client/ResultsMessage.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/client/src/main/java/org/teiid/client/ResultsMessage.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -26,7 +26,7 @@ import java.io.IOException; import java.io.ObjectInput; import java.io.ObjectOutput; -import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.List; @@ -45,9 +45,9 @@ static final long serialVersionUID = 3546924172976187793L; - private List[] results = null; - private String[] columnNames = null; - private String[] dataTypes = null; + private List<? extends List<?>> results; + private String[] columnNames; + private String[] dataTypes; /** A description of planning that occurred as requested in the request. */ private PlanNode planDescription; @@ -85,36 +85,29 @@ public ResultsMessage(){ } - /** - * Instantiate and copy relevant information from the original request message. - * Typically, the transaction context should only be copied if this results - * message is being returned from the connector to the query engine. Clients - * will be unable to deserialize this object. - * @param requestMsg - * @param copyTransactionContext true if the transaction context should be copied; false otherwise. - * @since 4.2 - */ - public ResultsMessage(RequestMessage requestMsg){ - this.results = new ArrayList[0]; - - } - - public ResultsMessage(RequestMessage requestMsg, List[] results, String[] columnNames, String[] dataTypes){ - this (requestMsg); - setResults( results ); + public ResultsMessage(List<? extends List<?>> results, String[] columnNames, String[] dataTypes){ + this.results = results; setFirstRow( 1 ); - setLastRow( results.length ); + setLastRow( results.size() ); this.columnNames = columnNames; this.dataTypes = dataTypes; } - - public List[] getResults() { + + public List<? extends List<?>> getResultsList() { return results; } + + /** + * @deprecated see {@link #getResultsList()} + * @return + */ + public List<?>[] getResults() { + return results.toArray(new List[results.size()]); + } - public void setResults(List[] results) { - this.results = results; + public void setResults(List<?>[] results) { + this.results = Arrays.asList(results); } public String[] getColumnNames() { @@ -341,7 +334,7 @@ */ public String toString() { return new StringBuffer("ResultsMessage rowCount=") //$NON-NLS-1$ - .append(results == null ? 0 : results.length) + .append(results == null ? 0 : results.size()) .append(" finalRow=") //$NON-NLS-1$ .append(finalRow) .toString(); Modified: trunk/client/src/main/java/org/teiid/jdbc/ResultSetImpl.java =================================================================== --- trunk/client/src/main/java/org/teiid/jdbc/ResultSetImpl.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/client/src/main/java/org/teiid/jdbc/ResultSetImpl.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -416,7 +416,7 @@ private Batch getCurrentBatch(ResultsMessage currentResultMsg) { this.updatedPlanDescription = currentResultMsg.getPlanDescription(); - boolean isLast = currentResultMsg.getResults().length == 0 || currentResultMsg.getFinalRow() == currentResultMsg.getLastRow(); + boolean isLast = currentResultMsg.getResultsList().size() == 0 || currentResultMsg.getFinalRow() == currentResultMsg.getLastRow(); Batch result = new Batch(currentResultMsg.getResults(), currentResultMsg.getFirstRow(), currentResultMsg.getLastRow(), isLast); result.setLastRow(currentResultMsg.getFinalRow()); return result; Modified: trunk/client/src/test/java/org/teiid/client/TestBatchSerializer.java =================================================================== --- trunk/client/src/test/java/org/teiid/client/TestBatchSerializer.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/client/src/test/java/org/teiid/client/TestBatchSerializer.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -35,10 +35,10 @@ import java.util.Arrays; import java.util.List; -import org.teiid.client.BatchSerializer; +import junit.framework.TestCase; + import org.teiid.core.types.DataTypeManager; -import junit.framework.TestCase; @@ -63,21 +63,22 @@ } } - private static void helpTestSerialization(String[] types, List[] batch) throws IOException, ClassNotFoundException { + private static void helpTestSerialization(String[] types, List<?>[] batch) throws IOException, ClassNotFoundException { ByteArrayOutputStream byteStream = new ByteArrayOutputStream(); ObjectOutputStream out = new ObjectOutputStream(byteStream); - BatchSerializer.writeBatch(out, types, batch); + List<List<?>> batchList = Arrays.asList(batch); + BatchSerializer.writeBatch(out, types, batchList); out.flush(); byte[] bytes = byteStream.toByteArray(); ByteArrayInputStream bytesIn = new ByteArrayInputStream(bytes); ObjectInputStream in = new ObjectInputStream(bytesIn); - List[] newBatch = BatchSerializer.readBatch(in, types); + List<List<Object>> newBatch = BatchSerializer.readBatch(in, types); out.close(); in.close(); - assertEqual(batch, newBatch); + assertTrue(batchList.equals(newBatch)); } private static final String[] sampleBatchTypes = {DataTypeManager.DefaultDataTypes.BIG_DECIMAL, @@ -164,14 +165,6 @@ helpTestSerialization(sampleBatchTypes, sampleBatch(833)); // A bunch of rows. This should also test large strings } - public void testSerializeBasicTypes_NoTypeHints() throws Exception { - helpTestSerialization(null, sampleBatch(1)); - helpTestSerialization(null, sampleBatch(8)); - helpTestSerialization(null, sampleBatch(17)); - helpTestSerialization(null, sampleBatch(120)); - helpTestSerialization(null, sampleBatch(833)); - } - public void testSerializeBasicTypesWithNulls() throws Exception { helpTestSerialization(sampleBatchTypes, sampleBatchWithNulls(1)); helpTestSerialization(sampleBatchTypes, sampleBatchWithNulls(8)); @@ -186,10 +179,7 @@ } public void testSerializeNoData() throws Exception { - helpTestSerialization(sampleBatchTypes, null); - helpTestSerialization(null, null); helpTestSerialization(sampleBatchTypes, new List[0]); - helpTestSerialization(null, new List[0]); } public void testSerializeDatatypeMismatch() throws Exception { Modified: trunk/client/src/test/java/org/teiid/jdbc/TestAllResultsImpl.java =================================================================== --- trunk/client/src/test/java/org/teiid/jdbc/TestAllResultsImpl.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/client/src/test/java/org/teiid/jdbc/TestAllResultsImpl.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -797,7 +797,7 @@ private ResultsMessage exampleMessage(List<Object>[] results, String[] columnNames, String[] datatypes) { RequestMessage request = new RequestMessage(); request.setExecutionId(REQUEST_ID); - ResultsMessage resultsMsg = new ResultsMessage(request); + ResultsMessage resultsMsg = new ResultsMessage(); resultsMsg.setResults(results); resultsMsg.setColumnNames(columnNames); resultsMsg.setDataTypes(datatypes); @@ -834,7 +834,7 @@ private static ResultsMessage exampleResultsMsg4(int begin, int length, boolean lastBatch) { RequestMessage request = new RequestMessage(); request.setExecutionId(REQUEST_ID); - ResultsMessage resultsMsg = new ResultsMessage(request); + ResultsMessage resultsMsg = new ResultsMessage(); List[] results = exampleResults1(length, begin); resultsMsg.setResults(results); resultsMsg.setColumnNames(new String[] { "IntKey" }); //$NON-NLS-1$ @@ -862,7 +862,7 @@ @Test public void testDateType() throws SQLException { RequestMessage request = new RequestMessage(); request.setExecutionId(REQUEST_ID); - ResultsMessage resultsMsg = new ResultsMessage(request); + ResultsMessage resultsMsg = new ResultsMessage(); resultsMsg.setResults(new List[] {Arrays.asList(new Timestamp(0))}); resultsMsg.setColumnNames(new String[] { "TS" }); //$NON-NLS-1$ resultsMsg.setDataTypes(new String[] { JDBCSQLTypeInfo.TIMESTAMP }); Modified: trunk/client/src/test/java/org/teiid/jdbc/TestCallableStatement.java =================================================================== --- trunk/client/src/test/java/org/teiid/jdbc/TestCallableStatement.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/client/src/test/java/org/teiid/jdbc/TestCallableStatement.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -38,9 +38,6 @@ import org.teiid.client.metadata.ParameterInfo; import org.teiid.client.security.LogonResult; import org.teiid.core.types.JDBCSQLTypeInfo; -import org.teiid.jdbc.CallableStatementImpl; -import org.teiid.jdbc.ConnectionImpl; -import org.teiid.jdbc.ResultSetImpl; import org.teiid.net.ServerConnection; @@ -68,7 +65,7 @@ RequestMessage request = new RequestMessage(); request.setExecutionId(1); - ResultsMessage resultsMsg = new ResultsMessage(request); + ResultsMessage resultsMsg = new ResultsMessage(); List[] results = new List[] {Arrays.asList(null, null, null), Arrays.asList(null, 1, 2)}; resultsMsg.setResults(results); resultsMsg.setColumnNames(new String[] { "IntNum", "Out1", "Out2" }); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ Modified: trunk/common-core/src/main/java/org/teiid/core/types/DataTypeManager.java =================================================================== --- trunk/common-core/src/main/java/org/teiid/core/types/DataTypeManager.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/common-core/src/main/java/org/teiid/core/types/DataTypeManager.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -126,23 +126,40 @@ } } - private static Map<Class<?>, ValueCache<?>> valueMaps = new HashMap<Class<?>, ValueCache<?>>(128); - private static HashedValueCache<String> stringCache = new HashedValueCache<String>(17) { + public static class WeakReferenceHashedValueCache<T> extends HashedValueCache<T> { + public WeakReferenceHashedValueCache(int size) { + super(size); + } + + public T getByHash(Object obj) { + int index = hash(obj.hashCode()) & (cache.length - 1); + return get(index); + } + @Override - protected Object get(int index) { - WeakReference<?> ref = (WeakReference<?>) cache[index]; + protected T get(int index) { + WeakReference<T> ref = (WeakReference<T>) cache[index]; if (ref != null) { - return ref.get(); + T result = ref.get(); + if (result == null) { + cache[index] = null; + } + return result; } return null; } @Override - protected void set(int index, String value) { - cache[index] = new WeakReference<Object>(value); - } + protected void set(int index, T value) { + cache[index] = new WeakReference<T>(value); + } + } + + private static Map<Class<?>, ValueCache<?>> valueMaps = new HashMap<Class<?>, ValueCache<?>>(128); + private static HashedValueCache<String> stringCache = new WeakReferenceHashedValueCache<String>(17) { + @Override protected int primaryHash(String value) { if (value.length() < 14) { @@ -536,21 +553,7 @@ valueMaps.put(DefaultDataClasses.DATE, new HashedValueCache<Date>(14)); valueMaps.put(DefaultDataClasses.TIME, new HashedValueCache<Time>(14)); valueMaps.put(DefaultDataClasses.TIMESTAMP, new HashedValueCache<Timestamp>(14)); - valueMaps.put(DefaultDataClasses.BIG_DECIMAL, new HashedValueCache<BigDecimal>(16) { - @Override - protected Object get(int index) { - WeakReference<?> ref = (WeakReference<?>) cache[index]; - if (ref != null) { - return ref.get(); - } - return null; - } - - @Override - protected void set(int index, BigDecimal value) { - cache[index] = new WeakReference<BigDecimal>(value); - } - }); + valueMaps.put(DefaultDataClasses.BIG_DECIMAL, new WeakReferenceHashedValueCache<BigDecimal>(16)); valueMaps.put(DefaultDataClasses.STRING, stringCache); } } Added: trunk/engine/src/main/java/org/teiid/common/buffer/AutoCleanupUtil.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/AutoCleanupUtil.java (rev 0) +++ trunk/engine/src/main/java/org/teiid/common/buffer/AutoCleanupUtil.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -0,0 +1,73 @@ +/* + * JBoss, Home of Professional Open Source. + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301 USA. + */ +package org.teiid.common.buffer; + +import java.lang.ref.PhantomReference; +import java.lang.ref.ReferenceQueue; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.Set; + +public class AutoCleanupUtil { + + public interface Removable { + public void remove(); + } + + static final class PhantomCleanupReference extends PhantomReference<Object> { + + private Removable removable; + + public PhantomCleanupReference(Object referent, Removable removable) { + super(referent, QUEUE); + this.removable = removable; + } + + public void cleanup() { + try { + this.removable.remove(); + } finally { + this.removable = null; + this.clear(); + } + } + } + + private static ReferenceQueue<Object> QUEUE = new ReferenceQueue<Object>(); + private static final Set<PhantomReference<Object>> REFERENCES = Collections.synchronizedSet(Collections.newSetFromMap(new IdentityHashMap<PhantomReference<Object>, Boolean>())); + + public static void setCleanupReference(Object o, Removable r) { + REFERENCES.add(new PhantomCleanupReference(o, r)); + doCleanup(); + } + + public static void doCleanup() { + for (int i = 0; i < 10; i++) { + PhantomCleanupReference ref = (PhantomCleanupReference)QUEUE.poll(); + if (ref == null) { + break; + } + ref.cleanup(); + REFERENCES.remove(ref); + } + } +} Property changes on: trunk/engine/src/main/java/org/teiid/common/buffer/AutoCleanupUtil.java ___________________________________________________________________ Added: svn:mime-type + text/plain Modified: trunk/engine/src/main/java/org/teiid/common/buffer/BatchManager.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/BatchManager.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/BatchManager.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -22,36 +22,29 @@ package org.teiid.common.buffer; +import java.lang.ref.Reference; +import java.util.List; + import org.teiid.core.TeiidComponentException; +/** + * Acts as a combination serializer/cachemanager + */ public interface BatchManager { - public interface CleanupHook { - - void cleanup(); - - } + List<List<?>> getBatch(Long batch, boolean retain) throws TeiidComponentException; - public interface ManagedBatch { - - TupleBatch getBatch(boolean cache, String[] types) throws TeiidComponentException; - - void remove(); - - void setPrefersMemory(boolean prefers); - - /** - * Get an object that can cleaup the {@link ManagedBatch}, but does not hold a hard reference to - * the {@link ManagedBatch} or the {@link BatchManager} - * @return - */ - CleanupHook getCleanupHook(); - - } + void remove(Long batch); - ManagedBatch createManagedBatch(TupleBatch batch, boolean softCache) throws TeiidComponentException; + void setPrefersMemory(boolean prefers); + boolean prefersMemory(); + + Long createManagedBatch(List<? extends List<?>> batch) throws TeiidComponentException; + void remove(); - FileStore createStorage(String prefix); + Reference<? extends BatchManager> getBatchManagerReference(); + + String[] getTypes(); } Added: trunk/engine/src/main/java/org/teiid/common/buffer/Cache.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/Cache.java (rev 0) +++ trunk/engine/src/main/java/org/teiid/common/buffer/Cache.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -0,0 +1,39 @@ +/* + * JBoss, Home of Professional Open Source. + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301 USA. + */ + +package org.teiid.common.buffer; + +import java.util.Collection; + +import org.teiid.core.TeiidComponentException; + +/** + * Represents the storage strategy for the {@link BufferManager} + */ +public interface Cache extends StorageManager { + void createCacheGroup(Long gid); + Collection<Long> removeCacheGroup(Long gid); + void addToCacheGroup(Long gid, Long oid); //called prior to adding an entry + CacheEntry get(Long id, Serializer<?> serializer) throws TeiidComponentException; + void add(CacheEntry entry, Serializer<?> s); + void remove(Long gid, Long id); +} \ No newline at end of file Property changes on: trunk/engine/src/main/java/org/teiid/common/buffer/Cache.java ___________________________________________________________________ Added: svn:mime-type + text/plain Added: trunk/engine/src/main/java/org/teiid/common/buffer/CacheEntry.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/CacheEntry.java (rev 0) +++ trunk/engine/src/main/java/org/teiid/common/buffer/CacheEntry.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -0,0 +1,101 @@ +/* + * JBoss, Home of Professional Open Source. + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301 USA. + */ + +package org.teiid.common.buffer; + +import java.lang.ref.WeakReference; + +public class CacheEntry { + private boolean persistent; + private Object object; + private int sizeEstimate; + private WeakReference<? extends Serializer<?>> serializer; + private Long id; + + public CacheEntry(Long id) { + this.id = id; + } + + public Long getId() { + return id; + } + + @Override + public int hashCode() { + return getId().hashCode(); + } + + public int getSizeEstimate() { + return sizeEstimate; + } + + public void setSizeEstimate(int sizeEstimate) { + this.sizeEstimate = sizeEstimate; + } + + public boolean equals(Object obj) { + if (obj == this) { + return true; + } + if (!(obj instanceof CacheEntry)) { + return false; + } + return getId().equals(((CacheEntry)obj).getId()); + } + + @Override + public String toString() { + return getId().toString(); + } + + public Object nullOut() { + Object result = getObject(); + setObject(null); + setSerializer(null); + return result; + } + + public void setObject(Object object) { + this.object = object; + } + + public Object getObject() { + return object; + } + + public void setPersistent(boolean persistent) { + this.persistent = persistent; + } + + public boolean isPersistent() { + return persistent; + } + + public void setSerializer(WeakReference<? extends Serializer<?>> serializer) { + this.serializer = serializer; + } + + public WeakReference<? extends Serializer<?>> getSerializer() { + return serializer; + } + +} \ No newline at end of file Property changes on: trunk/engine/src/main/java/org/teiid/common/buffer/CacheEntry.java ___________________________________________________________________ Added: svn:mime-type + text/plain Modified: trunk/engine/src/main/java/org/teiid/common/buffer/FileStore.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/FileStore.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/FileStore.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -25,20 +25,13 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; -import java.lang.ref.PhantomReference; -import java.lang.ref.ReferenceQueue; import java.util.Arrays; -import java.util.Collections; -import java.util.IdentityHashMap; -import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; -import org.teiid.core.TeiidComponentException; +import org.teiid.common.buffer.AutoCleanupUtil.Removable; -public abstract class FileStore { - - private static ReferenceQueue<Object> QUEUE = new ReferenceQueue<Object>(); - private static final Set<PhantomReference<Object>> REFERENCES = Collections.newSetFromMap(new IdentityHashMap<PhantomReference<Object>, Boolean>()); - +public abstract class FileStore implements Removable { + /** * A customized buffered stream with an exposed buffer */ @@ -80,12 +73,8 @@ } private void writeDirect(byte[] b, int off, int len) throws IOException { - try { - FileStore.this.write(b, off, len); - bytesWritten = true; - } catch (TeiidComponentException e) { - throw new IOException(e); - } + FileStore.this.write(b, off, len); + bytesWritten = true; } public void flushBuffer() throws IOException { @@ -140,91 +129,59 @@ } - static class CleanupReference extends PhantomReference<Object> { - - private FileStore store; - - public CleanupReference(Object referent, FileStore store) { - super(referent, QUEUE); - this.store = store; - } - - public void cleanup() { - try { - this.store.remove(); - } finally { - this.clear(); - } - } - } + private AtomicBoolean removed = new AtomicBoolean(); - private boolean removed; - protected long len; + public abstract long getLength(); - public void setCleanupReference(Object o) { - REFERENCES.add(new CleanupReference(o, this)); - for (int i = 0; i < 10; i++) { - CleanupReference ref = (CleanupReference)QUEUE.poll(); - if (ref == null) { - break; - } - ref.cleanup(); - REFERENCES.remove(ref); - } - } + public abstract void setLength(long length) throws IOException; - public synchronized long getLength() { - return len; - } - - public synchronized void truncate(long length) throws TeiidComponentException { - truncateDirect(length); - len = length; - } - - protected abstract void truncateDirect(long length) throws TeiidComponentException; - public int read(long fileOffset, byte[] b, int offSet, int length) - throws TeiidComponentException { - if (removed) { - throw new TeiidComponentException("already removed"); //$NON-NLS-1$ + throws IOException { + checkRemoved(); + return readWrite(fileOffset, b, offSet, length, false); + } + + private void checkRemoved() throws IOException { + if (removed.get()) { + throw new IOException("already removed"); //$NON-NLS-1$ } - return readDirect(fileOffset, b, offSet, length); } - protected abstract int readDirect(long fileOffset, byte[] b, int offSet, int length) - throws TeiidComponentException; + protected abstract int readWrite(long fileOffset, byte[] b, int offSet, int length, boolean write) + throws IOException; - public void readFully(long fileOffset, byte[] b, int offSet, int length) throws TeiidComponentException { + public void readFully(long fileOffset, byte[] b, int offSet, int length) throws IOException { + if (length == 0) { + return; + } int n = 0; do { int count = this.read(fileOffset + n, b, offSet + n, length - n); - if (count < 0) { - throw new TeiidComponentException("not enough bytes available"); //$NON-NLS-1$ + if (count <= 0 && length > 0) { + throw new IOException("not enough bytes available"); //$NON-NLS-1$ } n += count; } while (n < length); } - public synchronized long write(byte[] bytes, int offset, int length) throws TeiidComponentException { - return write(len, bytes, offset, length); + public synchronized void write(byte[] bytes, int offset, int length) throws IOException { + write(getLength(), bytes, offset, length); } - public synchronized long write(long start, byte[] bytes, int offset, int length) throws TeiidComponentException { - if (removed) { - throw new TeiidComponentException("already removed"); //$NON-NLS-1$ - } - writeDirect(start, bytes, offset, length); - long result = len; - len = Math.max(len, start + length); - return result; + public void write(long start, byte[] bytes, int offset, int length) throws IOException { + int n = 0; + do { + checkRemoved(); + int count = this.readWrite(start + n, bytes, offset + n, length - n, true); + if (count <= 0 && length > 0) { + throw new IOException("not enough bytes available"); //$NON-NLS-1$ + } + n += count; + } while (n < length); } - protected abstract void writeDirect(long start, byte[] bytes, int offset, int length) throws TeiidComponentException; - - public synchronized void remove() { - if (!this.removed) { - this.removed = true; + public void remove() { + if (removed.compareAndSet(false, true)) { this.removeDirect(); } } @@ -248,24 +205,20 @@ @Override public int read(byte[] b, int off, int len) throws IOException { - try { - if (this.streamLength != -1 && len > this.streamLength) { - len = (int)this.streamLength; - } - if (this.streamLength == -1 || this.streamLength > 0) { - int bytes = FileStore.this.read(offset, b, off, len); - if (bytes != -1) { - this.offset += bytes; - if (this.streamLength != -1) { - this.streamLength -= bytes; - } + if (this.streamLength != -1 && len > this.streamLength) { + len = (int)this.streamLength; + } + if (this.streamLength == -1 || this.streamLength > 0) { + int bytes = FileStore.this.read(offset, b, off, len); + if (bytes != -1) { + this.offset += bytes; + if (this.streamLength != -1) { + this.streamLength -= bytes; } - return bytes; } - return -1; - } catch (TeiidComponentException e) { - throw new IOException(e); + return bytes; } + return -1; } }; } @@ -284,11 +237,7 @@ @Override public void write(byte[] b, int off, int len) throws IOException { - try { - FileStore.this.write(b, off, len); - } catch (TeiidComponentException e) { - throw new IOException(e); - } + FileStore.this.write(b, off, len); } }; } Modified: trunk/engine/src/main/java/org/teiid/common/buffer/FileStoreInputStreamFactory.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/FileStoreInputStreamFactory.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/FileStoreInputStreamFactory.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -42,7 +42,7 @@ public FileStoreInputStreamFactory(FileStore lobBuffer, String encoding) { this.encoding = encoding; this.lobBuffer = lobBuffer; - this.lobBuffer.setCleanupReference(this); + AutoCleanupUtil.setCleanupReference(this, lobBuffer); } @Override Added: trunk/engine/src/main/java/org/teiid/common/buffer/LightWeightCopyOnWriteList.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/LightWeightCopyOnWriteList.java (rev 0) +++ trunk/engine/src/main/java/org/teiid/common/buffer/LightWeightCopyOnWriteList.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -0,0 +1,116 @@ +/* + * JBoss, Home of Professional Open Source. + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301 USA. + */ + +package org.teiid.common.buffer; + +import java.util.AbstractList; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.RandomAccess; + +/** + * Creates a copy of a reference list when modified. + * + * @param <T> + */ +public class LightWeightCopyOnWriteList<T> extends AbstractList<T> implements RandomAccess { + + private List<T> list; + private boolean modified; + + public LightWeightCopyOnWriteList(List<T> list) { + this.list = list; + } + + @Override + public T get(int index) { + return list.get(index); + } + + public List<T> getList() { + return list; + } + + public void add(int index, T element) { + if (!modified) { + List<T> next = new ArrayList<T>(list.size() + 1); + next.addAll(list); + list = next; + modified = true; + } + list.add(index, element); + } + + public T set(int index, T element) { + checkModified(); + return list.set(index, element); + } + + private void checkModified() { + if (!modified) { + list = new ArrayList<T>(list); + modified = true; + } + } + + public boolean addAll(Collection<? extends T> c) { + return addAll(size(), c); + } + + @Override + public boolean addAll(int index, Collection<? extends T> c) { + checkModified(); + return list.addAll(index, c); + } + + @Override + public T remove(int index) { + checkModified(); + return list.remove(index); + } + + @Override + public Object[] toArray() { + return list.toArray(); + } + + public <U extends Object> U[] toArray(U[] a) { + return list.toArray(a); + } + + @Override + public void clear() { + if (!modified) { + list = new ArrayList<T>(); + modified = true; + } else { + list.clear(); + } + } + + @Override + public int size() { + return list.size(); + } + +} Property changes on: trunk/engine/src/main/java/org/teiid/common/buffer/LightWeightCopyOnWriteList.java ___________________________________________________________________ Added: svn:mime-type + text/plain Modified: trunk/engine/src/main/java/org/teiid/common/buffer/LobManager.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/LobManager.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/LobManager.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -30,9 +30,10 @@ import java.sql.SQLException; import java.sql.SQLXML; import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; import org.teiid.core.TeiidComponentException; import org.teiid.core.types.BaseLob; @@ -58,10 +59,47 @@ * TODO: for temp tables we may need to have a copy by value management strategy */ public class LobManager { - private Map<String, Streamable<?>> lobReferences = new ConcurrentHashMap<String, Streamable<?>>(); + + public enum ReferenceMode { + ATTACH, + CREATE, + REMOVE + } + + private static class LobHolder { + Streamable<?> lob; + int referenceCount = 1; + + public LobHolder(Streamable<?> lob) { + this.lob = lob; + } + } + + private Map<String, LobHolder> lobReferences = Collections.synchronizedMap(new HashMap<String, LobHolder>()); private boolean inlineLobs = true; private int maxMemoryBytes = DataTypeManager.MAX_LOB_MEMORY_BYTES; + private int[] lobIndexes; + private FileStore lobStore; + public LobManager(int[] lobIndexes, FileStore lobStore) { + this.lobIndexes = lobIndexes; + this.lobStore = lobStore; + } + + public LobManager clone() { + LobManager clone = new LobManager(lobIndexes, null); + clone.inlineLobs = inlineLobs; + clone.maxMemoryBytes = maxMemoryBytes; + synchronized (lobReferences) { + for (Map.Entry<String, LobHolder> entry : lobReferences.entrySet()) { + LobHolder lobHolder = new LobHolder(entry.getValue().lob); + lobHolder.referenceCount = entry.getValue().referenceCount; + clone.lobReferences.put(entry.getKey(), lobHolder); + } + } + return clone; + } + public void setInlineLobs(boolean trackMemoryLobs) { this.inlineLobs = trackMemoryLobs; } @@ -69,8 +107,9 @@ public void setMaxMemoryBytes(int maxMemoryBytes) { this.maxMemoryBytes = maxMemoryBytes; } - - public void updateReferences(int[] lobIndexes, List<?> tuple) + + @SuppressWarnings("unchecked") + public void updateReferences(List<?> tuple, ReferenceMode mode) throws TeiidComponentException { for (int i = 0; i < lobIndexes.length; i++) { Object anObj = tuple.get(lobIndexes[i]); @@ -79,30 +118,53 @@ } Streamable lob = (Streamable) anObj; try { - if (inlineLobs + if (lob.getReferenceStreamId() == null || (inlineLobs && (InputStreamFactory.getStorageMode(lob) == StorageMode.MEMORY - || lob.length()*(lob instanceof ClobType?2:1) <= maxMemoryBytes)) { + || lob.length()*(lob instanceof ClobType?2:1) <= maxMemoryBytes))) { lob.setReferenceStreamId(null); continue; } } catch (SQLException e) { //presumably the lob is bad, but let it slide for now } - if (lob.getReference() == null) { - lob.setReference(getLobReference(lob.getReferenceStreamId()).getReference()); - } else { - String id = lob.getReferenceStreamId(); - this.lobReferences.put(id, lob); + String id = lob.getReferenceStreamId(); + LobHolder lobHolder = this.lobReferences.get(id); + switch (mode) { + case REMOVE: + if (lobHolder != null) { + lobHolder.referenceCount--; + if (lobHolder.referenceCount < 1) { + this.lobReferences.remove(id); + } + } + break; + case ATTACH: + if (lob.getReference() == null) { + if (lobHolder == null) { + throw new TeiidComponentException(QueryPlugin.Util.getString("ProcessWorker.wrongdata")); //$NON-NLS-1$ + } + lob.setReference(lobHolder.lob.getReference()); + } + break; + case CREATE: + if (lob.getReference() == null) { + throw new TeiidComponentException(QueryPlugin.Util.getString("ProcessWorker.wrongdata")); //$NON-NLS-1$ + } + if (lobHolder == null) { + this.lobReferences.put(id, new LobHolder(lob)); + } else { + lobHolder.referenceCount++; + } } } } public Streamable<?> getLobReference(String id) throws TeiidComponentException { - Streamable<?> lob = this.lobReferences.get(id); + LobHolder lob = this.lobReferences.get(id); if (lob == null) { throw new TeiidComponentException(QueryPlugin.Util.getString("ProcessWorker.wrongdata")); //$NON-NLS-1$ } - return lob; + return lob.lob; } public static int[] getLobIndexes(List<? extends Expression> expressions) { @@ -123,12 +185,12 @@ return Arrays.copyOf(result, resultIndex); } - public void persist(FileStore lobStore) throws TeiidComponentException { + public void persist() throws TeiidComponentException { // stream the contents of lob into file store. - byte[] bytes = new byte[102400]; // 100k + byte[] bytes = new byte[1 << 14]; - for (Map.Entry<String, Streamable<?>> entry : this.lobReferences.entrySet()) { - entry.setValue(persistLob(entry.getValue(), lobStore, bytes)); + for (Map.Entry<String, LobHolder> entry : this.lobReferences.entrySet()) { + entry.getValue().lob = persistLob(entry.getValue().lob, lobStore, bytes); } } @@ -202,4 +264,9 @@ public int getLobCount() { return this.lobReferences.size(); } + + public void remove() { + this.lobReferences.clear(); + + } } Modified: trunk/engine/src/main/java/org/teiid/common/buffer/SPage.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/SPage.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/SPage.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -23,6 +23,7 @@ package org.teiid.common.buffer; import java.lang.ref.PhantomReference; +import java.lang.ref.Reference; import java.lang.ref.ReferenceQueue; import java.util.ArrayList; import java.util.Collections; @@ -33,16 +34,12 @@ import java.util.Set; import java.util.concurrent.atomic.AtomicLong; -import org.teiid.common.buffer.BatchManager.CleanupHook; -import org.teiid.common.buffer.BatchManager.ManagedBatch; import org.teiid.core.TeiidComponentException; import org.teiid.core.TeiidRuntimeException; /** * A linked list Page entry in the tree * - * TODO: return the tuplebatch from getvalues, since that is what we're tracking - * * State cloning allows a single storage reference to be shared in many trees. * A phantom reference is used for proper cleanup once cloned. * @@ -57,8 +54,8 @@ static class SearchResult { int index; SPage page; - TupleBatch values; - public SearchResult(int index, SPage page, TupleBatch values) { + List<List<?>> values; + public SearchResult(int index, SPage page, List<List<?>> values) { this.index = index; this.page = page; this.values = values; @@ -69,16 +66,21 @@ private static ReferenceQueue<Object> QUEUE = new ReferenceQueue<Object>(); static class CleanupReference extends PhantomReference<Object> { - private CleanupHook batch; + private Long batch; + private Reference<? extends BatchManager> ref; - public CleanupReference(Object referent, CleanupHook batch) { + public CleanupReference(Object referent, Long batch, Reference<? extends BatchManager> ref) { super(referent, QUEUE); this.batch = batch; + this.ref = ref; } public void cleanup() { try { - this.batch.cleanup(); + BatchManager batchManager = ref.get(); + if (batchManager != null) { + batchManager.remove(batch); + } } finally { this.clear(); } @@ -92,28 +94,26 @@ private long id; protected SPage next; protected SPage prev; - protected ManagedBatch managedBatch; - private Object trackingObject; - protected TupleBatch values; - protected ArrayList<SPage> children; - protected boolean cloned; + protected Long managedBatch; + protected Object trackingObject; + protected List<List<?>> values; + protected List<SPage> children; SPage(STree stree, boolean leaf) { this.stree = stree; this.id = counter.getAndIncrement(); stree.pages.put(this.id, this); - this.values = new TupleBatch(0, new ArrayList(stree.getPageSize(leaf)/4)); + this.values = new ArrayList<List<?>>(); if (!leaf) { - children = new ArrayList<SPage>(stree.getPageSize(false)/4); + children = new ArrayList<SPage>(); } } public SPage clone(STree tree) { try { if (this.managedBatch != null && trackingObject == null) { - cloned = true; this.trackingObject = new Object(); - CleanupReference managedBatchReference = new CleanupReference(trackingObject, managedBatch.getCleanupHook()); + CleanupReference managedBatchReference = new CleanupReference(trackingObject, managedBatch, stree.getBatchManager(children == null).getBatchManagerReference()); REFERENCES.add(managedBatchReference); } SPage clone = (SPage) super.clone(); @@ -122,7 +122,7 @@ clone.children = new ArrayList<SPage>(children); } if (values != null) { - clone.values = new TupleBatch(0, new ArrayList<List<?>>(values.getTuples())); + clone.values = new ArrayList<List<?>>(values); } return clone; } catch (CloneNotSupportedException e) { @@ -135,32 +135,32 @@ } static SearchResult search(SPage page, List k, LinkedList<SearchResult> parent) throws TeiidComponentException { - TupleBatch previousValues = null; + List<List<?>> previousValues = null; for (;;) { - TupleBatch values = page.getValues(); - int index = Collections.binarySearch(values.getTuples(), k, page.stree.comparator); + List<List<?>> values = page.getValues(); + int index = Collections.binarySearch(values, k, page.stree.comparator); int flippedIndex = - index - 1; if (previousValues != null) { if (flippedIndex == 0) { //systemic weakness of the algorithm - return new SearchResult(-previousValues.getTuples().size() - 1, page.prev, previousValues); + return new SearchResult(-previousValues.size() - 1, page.prev, previousValues); } if (parent != null && index != 0) { page.stree.updateLock.lock(); try { - index = Collections.binarySearch(values.getTuples(), k, page.stree.comparator); + index = Collections.binarySearch(values, k, page.stree.comparator); if (index != 0) { //for non-matches move the previous pointer over to this page SPage childPage = page; List oldKey = null; - List newKey = page.stree.extractKey(values.getTuples().get(0)); + List newKey = page.stree.extractKey(values.get(0)); for (Iterator<SearchResult> desc = parent.descendingIterator(); desc.hasNext();) { SearchResult sr = desc.next(); int parentIndex = Math.max(0, -sr.index - 2); if (oldKey == null) { - oldKey = sr.values.getTuples().set(parentIndex, newKey); - } else if (page.stree.comparator.compare(oldKey, sr.values.getTuples().get(parentIndex)) == 0 ) { - sr.values.getTuples().set(parentIndex, newKey); + oldKey = sr.values.set(parentIndex, newKey); + } else if (page.stree.comparator.compare(oldKey, sr.values.get(parentIndex)) == 0 ) { + sr.values.set(parentIndex, newKey); } else { break; } @@ -174,7 +174,7 @@ } } } - if (flippedIndex != values.getTuples().size() || page.next == null) { + if (flippedIndex != values.size() || page.next == null) { return new SearchResult(index, page, values); } previousValues = values; @@ -182,35 +182,32 @@ } } - protected void setValues(TupleBatch values) throws TeiidComponentException { - if (managedBatch != null && !cloned) { - managedBatch.remove(); + protected void setValues(List<List<?>> values) throws TeiidComponentException { + if (values instanceof LightWeightCopyOnWriteList<?>) { + values = ((LightWeightCopyOnWriteList<List<?>>)values).getList(); } - if (values.getTuples().size() < MIN_PERSISTENT_SIZE) { + if (managedBatch != null && trackingObject == null) { + stree.getBatchManager(children == null).remove(managedBatch); + managedBatch = null; + trackingObject = null; + } + if (values.size() < MIN_PERSISTENT_SIZE) { this.values = values; return; - } - this.values = null; - if (children != null) { - values.setDataTypes(stree.keytypes); - } else { - values.setDataTypes(stree.types); + } else if (stree.batchInsert && children == null && values.size() < stree.leafSize) { + this.values = values; + stree.incompleteInsert = this; + return; } - if (cloned) { - cloned = false; - trackingObject = null; - } - if (children != null) { - managedBatch = stree.keyManager.createManagedBatch(values, true); - } else { - managedBatch = stree.leafManager.createManagedBatch(values, stree.preferMemory); - } + this.values = null; + this.trackingObject = null; + managedBatch = stree.getBatchManager(children == null).createManagedBatch(values); } - + protected void remove(boolean force) { if (managedBatch != null) { - if (force || !cloned) { - managedBatch.remove(); + if (force || trackingObject == null) { + stree.getBatchManager(children == null).remove(managedBatch); } managedBatch = null; trackingObject = null; @@ -219,7 +216,7 @@ children = null; } - protected TupleBatch getValues() throws TeiidComponentException { + protected List<List<?>> getValues() throws TeiidComponentException { if (values != null) { return values; } @@ -234,19 +231,20 @@ REFERENCES.remove(ref); ref.cleanup(); } - if (children != null) { - return managedBatch.getBatch(true, stree.keytypes); + List<List<?>> result = stree.getBatchManager(children == null).getBatch(managedBatch, true); + if (trackingObject != null) { + return new LightWeightCopyOnWriteList<List<?>>(result); } - return managedBatch.getBatch(true, stree.types); + return result; } - static void merge(LinkedList<SearchResult> places, TupleBatch nextValues, SPage current, TupleBatch currentValues) + static void merge(LinkedList<SearchResult> places, List<List<?>> nextValues, SPage current, List<List<?>> currentValues) throws TeiidComponentException { SearchResult parent = places.peekLast(); if (parent != null) { - correctParents(parent.page, nextValues.getTuples().get(0), current.next, current); + correctParents(parent.page, nextValues.get(0), current.next, current); } - currentValues.getTuples().addAll(nextValues.getTuples()); + currentValues.addAll(nextValues); if (current.children != null) { current.children.addAll(current.next.children); } @@ -292,19 +290,19 @@ public String toString() { StringBuilder result = new StringBuilder(); try { - TupleBatch tb = getValues(); - result.append(tb.getBeginRow()); + List<List<?>> tb = getValues(); + result.append(id); if (children == null) { - if (tb.getTuples().size() <= 1) { - result.append(tb.getTuples()); + if (tb.size() <= 1) { + result.append(tb); } else { - result.append("[").append(tb.getTuples().get(0)).append(" . ").append(tb.getTuples().size()). //$NON-NLS-1$ //$NON-NLS-2$ - append(" . ").append(tb.getTuples().get(tb.getTuples().size() - 1)).append("]"); //$NON-NLS-1$ //$NON-NLS-2$ + result.append("[").append(tb.get(0)).append(" . ").append(tb.size()). //$NON-NLS-1$ //$NON-NLS-2$ + append(" . ").append(tb.get(tb.size() - 1)).append("]"); //$NON-NLS-1$ //$NON-NLS-2$ } } else { result.append("["); //$NON-NLS-1$ for (int i = 0; i < children.size(); i++) { - result.append(tb.getTuples().get(i)).append("->").append(children.get(i).getValues().getBeginRow()); //$NON-NLS-1$ + result.append(tb.get(i)).append("->").append(children.get(i).getId()); //$NON-NLS-1$ if (i < children.size() - 1) { result.append(", "); //$NON-NLS-1$ } Modified: trunk/engine/src/main/java/org/teiid/common/buffer/STree.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/STree.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/STree.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -27,15 +27,16 @@ import java.io.ObjectOutputStream; import java.util.ArrayList; import java.util.Arrays; +import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Random; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.ReentrantLock; import org.teiid.client.BatchSerializer; +import org.teiid.common.buffer.LobManager.ReferenceMode; import org.teiid.common.buffer.SPage.SearchResult; import org.teiid.core.TeiidComponentException; import org.teiid.core.TeiidRuntimeException; @@ -51,13 +52,13 @@ public enum InsertMode {ORDERED, NEW, UPDATE} - private static final Random seedGenerator = new Random(); + private static final Random seedGenerator = new Random(0); protected int randomSeed; private int mask = 1; private int shift = 1; - protected ConcurrentHashMap<Long, SPage> pages = new ConcurrentHashMap<Long, SPage>(); + protected HashMap<Long, SPage> pages = new HashMap<Long, SPage>(); protected volatile SPage[] header = new SPage[] {new SPage(this, true)}; protected BatchManager keyManager; protected BatchManager leafManager; @@ -65,9 +66,9 @@ private int pageSize; protected int leafSize; protected int keyLength; - protected String[] types; - protected String[] keytypes; - protected boolean preferMemory; + protected boolean batchInsert; + protected SPage incompleteInsert; + protected LobManager lobManager; protected ReentrantLock updateLock = new ReentrantLock(); @@ -79,9 +80,10 @@ int pageSize, int leafSize, int keyLength, - String[] types) { + LobManager lobManager) { randomSeed = seedGenerator.nextInt() | 0x00000100; // ensure nonzero this.keyManager = manager; + manager.setPrefersMemory(true); this.leafManager = leafManager; this.comparator = comparator; this.pageSize = Math.max(pageSize, SPage.MIN_PERSISTENT_SIZE); @@ -94,18 +96,20 @@ } this.leafSize = leafSize; this.keyLength = keyLength; - this.types = types; - this.keytypes = Arrays.copyOf(types, keyLength); + this.lobManager = lobManager; } public STree clone() { updateLock.lock(); try { STree clone = (STree) super.clone(); + if (lobManager != null) { + clone.lobManager = lobManager.clone(); + } clone.updateLock = new ReentrantLock(); clone.rowCount = new AtomicInteger(rowCount.get()); //clone the pages - clone.pages = new ConcurrentHashMap<Long, SPage>(pages); + clone.pages = new HashMap<Long, SPage>(pages); for (Map.Entry<Long, SPage> entry : clone.pages.entrySet()) { entry.setValue(entry.getValue().clone(clone)); } @@ -143,23 +147,41 @@ SPage page = header[0]; oos.writeInt(this.rowCount.get()); while (true) { - TupleBatch batch = page.getValues(); - BatchSerializer.writeBatch(oos, types, batch.getAllTuples()); + List<List<?>> batch = page.getValues(); + BatchSerializer.writeBatch(oos, leafManager.getTypes(), batch); if (page.next == null) { break; } } } + public void setBatchInsert(boolean batchInsert) throws TeiidComponentException { + if (this.batchInsert == batchInsert) { + return; + } + this.batchInsert = batchInsert; + if (batchInsert || incompleteInsert == null) { + return; + } + SPage toFlush = incompleteInsert; + incompleteInsert = null; + if (toFlush.managedBatch != null) { + return; + } + toFlush.setValues(toFlush.getValues()); + } + public void readValuesFrom(ObjectInputStream ois) throws IOException, ClassNotFoundException, TeiidComponentException { int size = ois.readInt(); int sizeHint = this.getExpectedHeight(size); + batchInsert = true; while (this.getRowCount() < size) { - List[] batch = BatchSerializer.readBatch(ois, types); + List<List<Object>> batch = BatchSerializer.readBatch(ois, leafManager.getTypes()); for (List list : batch) { this.insert(list, InsertMode.ORDERED, sizeHint); } } + batchInsert = false; } protected SPage findChildTail(SPage page) { @@ -221,7 +243,7 @@ if (places != null) { places.add(s); } - if ((s.index == -1 && s.page == header[i]) || s.values.getTuples().isEmpty()) { + if ((s.index == -1 && s.page == header[i]) || s.values.isEmpty()) { x = null; continue; //start at the beginning of the next level } @@ -236,7 +258,7 @@ if (!matched) { return null; } - return s.values.getTuples().get(index); + return s.values.get(index); } x = x.children.get(index); } @@ -250,13 +272,16 @@ public List insert(List tuple, InsertMode mode, int sizeHint) throws TeiidComponentException { LinkedList<SearchResult> places = new LinkedList<SearchResult>(); List match = null; + if (this.lobManager != null) { + this.lobManager.updateReferences(tuple, ReferenceMode.CREATE); + } if (mode == InsertMode.ORDERED) { SPage last = null; while (last == null || last.children != null) { last = findChildTail(last); //TODO: do this lazily - TupleBatch batch = last.getValues(); - places.add(new SearchResult(-batch.getTuples().size() -1, last, batch)); + List<List<?>> batch = last.getValues(); + places.add(new SearchResult(-batch.size() -1, last, batch)); } } else { match = find(tuple, places); @@ -266,8 +291,11 @@ } SearchResult last = places.getLast(); SPage page = last.page; - last.values.getTuples().set(last.index, tuple); + last.values.set(last.index, tuple); page.setValues(last.values); + if (this.lobManager != null) { + this.lobManager.updateReferences(tuple, ReferenceMode.REMOVE); + } return match; } } @@ -279,7 +307,7 @@ } else { level = randomLevel(); } - } else if (!places.isEmpty() && places.getLast().values.getTuples().size() == getPageSize(true)) { + } else if (!places.isEmpty() && places.getLast().values.size() == getPageSize(true)) { int row = rowCount.get(); while (row != 0 && row%getPageSize(true) == 0) { row = (row - getPageSize(true) + 1)/getPageSize(true); @@ -295,8 +323,8 @@ for (int i = 0; i <= level; i++) { if (places.isEmpty()) { SPage newHead = new SPage(this, false); - TupleBatch batch = newHead.getValues(); - batch.getTuples().add(key); + List<List<?>> batch = newHead.getValues(); + batch.add(key); newHead.setValues(batch); newHead.children.add(page); header[i] = newHead; @@ -333,9 +361,9 @@ SPage page = result.page; int index = -result.index - 1; boolean leaf = !(value instanceof SPage); - if (result.values.getTuples().size() == getPageSize(leaf)) { + if (result.values.size() == getPageSize(leaf)) { SPage nextPage = new SPage(this, leaf); - TupleBatch nextValues = nextPage.getValues(); + List<List<?>> nextValues = nextPage.getValues(); nextPage.next = page.next; nextPage.prev = page; if (nextPage.next != null) { @@ -345,8 +373,8 @@ boolean inNext = false; if (!ordered) { //split the values - nextValues.getTuples().addAll(result.values.getTuples().subList(getPageSize(leaf)/2, getPageSize(leaf))); - result.values.getTuples().subList(getPageSize(leaf)/2, getPageSize(leaf)).clear(); + nextValues.addAll(result.values.subList(getPageSize(leaf)/2, getPageSize(leaf))); + result.values.subList(getPageSize(leaf)/2, getPageSize(leaf)).clear(); if (!leaf) { nextPage.children.addAll(page.children.subList(getPageSize(leaf)/2, getPageSize(false))); page.children.subList(getPageSize(false)/2, getPageSize(false)).clear(); @@ -359,7 +387,7 @@ } page.setValues(result.values); if (parent != null) { - List min = nextPage.getValues().getTuples().get(0); + List min = nextPage.getValues().get(0); SPage.correctParents(parent.page, min, page, nextPage); } } else { @@ -377,12 +405,12 @@ return page; } - static void setValue(int index, List key, Object value, TupleBatch values, SPage page) { + static void setValue(int index, List key, Object value, List<List<?>> values, SPage page) { if (value instanceof SPage) { - values.getTuples().add(index, key); + values.add(index, key); page.children.add(index, (SPage) value); } else { - values.getTuples().add(index, (List)value); + values.add(index, (List)value); } } @@ -398,13 +426,13 @@ if (searchResult.index < 0) { continue; } - searchResult.values.getTuples().remove(searchResult.index); + searchResult.values.remove(searchResult.index); boolean leaf = true; if (searchResult.page.children != null) { leaf = false; searchResult.page.children.remove(searchResult.index); } - int size = searchResult.values.getTuples().size(); + int size = searchResult.values.size(); if (size == 0) { if (header[i] != searchResult.page) { searchResult.page.remove(false); @@ -431,15 +459,15 @@ } else if (size < getPageSize(leaf)/2) { //check for merge if (searchResult.page.next != null) { - TupleBatch nextValues = searchResult.page.next.getValues(); - if (nextValues.getTuples().size() < getPageSize(leaf)/4) { + List<List<?>> nextValues = searchResult.page.next.getValues(); + if (nextValues.size() < getPageSize(leaf)/4) { SPage.merge(places, nextValues, searchResult.page, searchResult.values); continue; } } if (searchResult.page.prev != null) { - TupleBatch prevValues = searchResult.page.prev.getValues(); - if (prevValues.getTuples().size() < getPageSize(leaf)/4) { + List<List<?>> prevValues = searchResult.page.prev.getValues(); + if (prevValues.size() < getPageSize(leaf)/4) { SPage.merge(places, searchResult.values, searchResult.page.prev, prevValues); continue; } @@ -447,6 +475,9 @@ } searchResult.page.setValues(searchResult.values); } + if (lobManager != null) { + lobManager.updateReferences(tuple, ReferenceMode.REMOVE); + } return tuple; } @@ -498,11 +529,11 @@ } public void setPreferMemory(boolean preferMemory) { - this.preferMemory = preferMemory; + this.leafManager.setPrefersMemory(preferMemory); } public boolean isPreferMemory() { - return preferMemory; + return this.leafManager.prefersMemory(); } public ListNestedSortComparator getComparator() { @@ -536,7 +567,7 @@ public void clearClonedFlags() { for (SPage page : pages.values()) { - page.cloned = false; + page.trackingObject = null; //we don't really care about using synchronization or a volatile here //since the worst case is that we'll just use gc cleanup } @@ -548,5 +579,12 @@ } return pageSize; } + + BatchManager getBatchManager(boolean leaf) { + if (leaf) { + return leafManager; + } + return keyManager; + } } \ No newline at end of file Added: trunk/engine/src/main/java/org/teiid/common/buffer/Serializer.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/Serializer.java (rev 0) +++ trunk/engine/src/main/java/org/teiid/common/buffer/Serializer.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -0,0 +1,38 @@ +/* + * JBoss, Home of Professional Open Source. + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301 USA. + */ + +package org.teiid.common.buffer; + +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; + +/** + * Responsible for serializing {@link CacheEntry}s + * @param <T> + */ +public interface Serializer<T> { + void serialize(T obj, ObjectOutputStream oos) throws IOException; + T deserialize(ObjectInputStream ois) throws IOException, ClassNotFoundException; + boolean useSoftCache(); + Long getId(); +} \ No newline at end of file Property changes on: trunk/engine/src/main/java/org/teiid/common/buffer/Serializer.java ___________________________________________________________________ Added: svn:mime-type + text/plain Modified: trunk/engine/src/main/java/org/teiid/common/buffer/TupleBatch.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/TupleBatch.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/TupleBatch.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -22,17 +22,11 @@ package org.teiid.common.buffer; -import java.io.Externalizable; -import java.io.IOException; -import java.io.ObjectInput; -import java.io.ObjectOutput; import java.util.ArrayList; import java.util.Arrays; import java.util.List; -import org.teiid.client.BatchSerializer; - /** * Represents a set of indexed tuples. The {@link #getBeginRow beginning row} * is the first row contained in this batch; if it equals "1" then it is the @@ -40,28 +34,17 @@ * tuples. The {@link #getEndRow ending row} is the last row contained in * this tuple batch; it is equal to the beginning row plus the * {@link #getRowCount number of rows} contained in this batch, minus one. - * This object is immutable and Serializable; */ -public class TupleBatch implements Externalizable { +public class TupleBatch { private static final long serialVersionUID = 6304443387337336957L; private int rowOffset; - private List<List<?>> tuples; + protected List<List<?>> tuples; // Optional state private boolean terminationFlag = false; - // for distributed cache purposes - private String[] preservedTypes; - - /** - * Contains ordered data types of each of the columns in the batch. Although it is not serialized, - * this array is a serialization aid and must be set before serialization and deserialization using - * the setDataTypes method. - */ - private transient String[] types; - /** Required to honor Externalizable contract */ public TupleBatch() { } @@ -152,14 +135,6 @@ this.terminationFlag = terminationFlag; } - public void setDataTypes(String[] types) { - this.types = types; - } - - public String[] getDataTypes() { - return types; - } - public boolean containsRow(int row) { return rowOffset <= row && getEndRow() >= row; } @@ -179,31 +154,8 @@ return s.toString(); } - public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { - rowOffset = in.readInt(); - terminationFlag = in.readBoolean(); - preservedTypes = (String[])in.readObject(); - if (types == null) { - types = preservedTypes; - } - tuples = new ArrayList<List<?>>(); - for (List tuple : BatchSerializer.readBatch(in, types)) { - tuples.add(tuple); - } - } - public void writeExternal(ObjectOutput out) throws IOException { - out.writeInt(this.rowOffset); - out.writeBoolean(terminationFlag); - out.writeObject(this.preservedTypes); - BatchSerializer.writeBatch(out, types, getAllTuples()); - } - public void setRowOffset(int rowOffset) { this.rowOffset = rowOffset; } - - public void preserveTypes() { - this.preservedTypes = types; - } } Modified: trunk/engine/src/main/java/org/teiid/common/buffer/TupleBrowser.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/TupleBrowser.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/TupleBrowser.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -48,7 +48,7 @@ private SPage bound; private int boundIndex = -1; - private TupleBatch values; + private List<List<?>> values; private boolean updated; private boolean direction; @@ -106,7 +106,7 @@ if (boundIndex < 0) { //we are guaranteed by find to not get back the -1 index, unless //there are no tuples, in which case a bound of -1 is fine - boundIndex = Math.min(upper.values.getTuples().size(), -boundIndex -1) - 1; + boundIndex = Math.min(upper.values.size(), -boundIndex -1) - 1; } if (!direction) { values = upper.values; @@ -122,7 +122,7 @@ if (page != bound || values == null) { values = bound.getValues(); } - boundIndex = values.getTuples().size() - 1; + boundIndex = values.size() - 1; } } @@ -173,14 +173,14 @@ continue; } if (values != null) { - int possibleIndex = Collections.binarySearch(values.getTuples(), newValue, tree.comparator); + int possibleIndex = Collections.binarySearch(values, newValue, tree.comparator); if (possibleIndex >= 0) { //value exists in the current page index = possibleIndex; - return values.getTuples().get(possibleIndex); + return values.get(possibleIndex); } //check for end/terminal conditions - if (direction && possibleIndex == -values.getTuples().size() -1) { + if (direction && possibleIndex == -values.size() -1) { if (page.next == null) { resetState(); return null; @@ -199,7 +199,7 @@ if (!setPage(newValue)) { continue; } - return values.getTuples().get(index); + return values.get(index); } if (page == null) { if (inPartial) { @@ -213,11 +213,11 @@ if (direction) { index = 0; } else { - index = values.getTuples().size() - 1; + index = values.size() - 1; } } - if (index >= 0 && index < values.getTuples().size()) { - List<?> result = values.getTuples().get(index); + if (index >= 0 && index < values.size()) { + List<?> result = values.get(index); if (page == bound && index == boundIndex) { resetState(); page = null; //terminate @@ -257,7 +257,7 @@ * @throws TeiidComponentException */ public void update(List<?> tuple) throws TeiidComponentException { - values.getTuples().set(index - getOffset(), tuple); + values.set(index - getOffset(), tuple); updated = true; } Modified: trunk/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -28,7 +28,7 @@ import java.util.Map; import java.util.TreeMap; -import org.teiid.common.buffer.BatchManager.ManagedBatch; +import org.teiid.common.buffer.LobManager.ReferenceMode; import org.teiid.core.TeiidComponentException; import org.teiid.core.types.DataTypeManager; import org.teiid.core.types.Streamable; @@ -64,33 +64,23 @@ private BatchManager manager; private String tupleSourceID; private List<? extends Expression> schema; - private String[] types; private int batchSize; private int rowCount; private boolean isFinal; - private TreeMap<Integer, BatchManager.ManagedBatch> batches = new TreeMap<Integer, BatchManager.ManagedBatch>(); + private TreeMap<Integer, Long> batches = new TreeMap<Integer, Long>(); private ArrayList<List<?>> batchBuffer; private boolean removed; private boolean forwardOnly; - private boolean prefersMemory; private LobManager lobManager; - private int[] lobIndexes; private String uuid; - private FileStore lobStore; - public TupleBuffer(BatchManager manager, String id, List<? extends Expression> schema, int[] lobIndexes, int batchSize) { + public TupleBuffer(BatchManager manager, String id, List<? extends Expression> schema, LobManager lobManager, int batchSize) { this.manager = manager; this.tupleSourceID = id; this.schema = schema; - this.types = getTypeNames(schema); - this.lobIndexes = lobIndexes; - if (this.lobIndexes != null) { - this.lobManager = new LobManager(); - this.lobStore = this.manager.createStorage("_lobs"); //$NON-NLS-1$ - this.lobStore.setCleanupReference(this); - } + this.lobManager = lobManager; this.batchSize = batchSize; } @@ -112,12 +102,12 @@ } public boolean isLobs() { - return lobIndexes != null; + return lobManager != null; } public void addTuple(List<?> tuple) throws TeiidComponentException { if (isLobs()) { - lobManager.updateReferences(lobIndexes, tuple); + lobManager.updateReferences(tuple, ReferenceMode.CREATE); } this.rowCount++; if (batchBuffer == null) { @@ -125,7 +115,7 @@ } batchBuffer.add(tuple); if (batchBuffer.size() == batchSize) { - saveBatch(false, false); + saveBatch(false); } } @@ -144,7 +134,7 @@ //add the lob references only, since they may still be referenced later if (isLobs()) { for (List<?> tuple : batch.getTuples()) { - lobManager.updateReferences(lobIndexes, tuple); + lobManager.updateReferences(tuple, ReferenceMode.CREATE); } } } @@ -154,7 +144,7 @@ throws TeiidComponentException { assert this.rowCount <= rowCount; if (this.rowCount != rowCount) { - saveBatch(false, true); + saveBatch(true); this.rowCount = rowCount; } } @@ -163,15 +153,15 @@ if (this.batchBuffer != null) { this.batchBuffer.clear(); } - for (BatchManager.ManagedBatch batch : this.batches.values()) { - batch.remove(); + for (Long batch : this.batches.values()) { + this.manager.remove(batch); } this.batches.clear(); } public void persistLobs() throws TeiidComponentException { if (this.lobManager != null) { - this.lobManager.persist(this.lobStore); + this.lobManager.persist(); } } @@ -180,26 +170,21 @@ * @throws TeiidComponentException */ public void saveBatch() throws TeiidComponentException { - this.saveBatch(false, false); + this.saveBatch(false); } - void saveBatch(boolean finalBatch, boolean force) throws TeiidComponentException { + void saveBatch(boolean force) throws TeiidComponentException { Assertion.assertTrue(!this.isRemoved()); if (batchBuffer == null || batchBuffer.isEmpty() || (!force && batchBuffer.size() < Math.max(1, batchSize / 32))) { return; } - TupleBatch writeBatch = new TupleBatch(rowCount - batchBuffer.size() + 1, batchBuffer); - if (finalBatch) { - writeBatch.setTerminationFlag(true); - } - writeBatch.setDataTypes(types); - BatchManager.ManagedBatch mbatch = manager.createManagedBatch(writeBatch, prefersMemory); - this.batches.put(writeBatch.getBeginRow(), mbatch); + Long mbatch = manager.createManagedBatch(batchBuffer); + this.batches.put(rowCount - batchBuffer.size() + 1, mbatch); batchBuffer = null; } public void close() throws TeiidComponentException { - saveBatch(true, false); + saveBatch(false); this.isFinal = true; } @@ -210,6 +195,8 @@ * @param row * @return * @throws TeiidComponentException + * + * TODO: a method to get the raw batch */ public TupleBatch getBatch(int row) throws TeiidComponentException { TupleBatch result = null; @@ -224,17 +211,20 @@ if (this.batchBuffer != null && !this.batchBuffer.isEmpty()) { //this is just a sanity check to ensure we're not holding too many //hard references to batches. - saveBatch(isFinal, false); + saveBatch(false); } - Map.Entry<Integer, BatchManager.ManagedBatch> entry = batches.floorEntry(row); + Map.Entry<Integer, Long> entry = batches.floorEntry(row); Assertion.isNotNull(entry); - BatchManager.ManagedBatch batch = entry.getValue(); - result = batch.getBatch(!forwardOnly, types); + Long batch = entry.getValue(); + List<List<?>> rows = manager.getBatch(batch, !forwardOnly); + result = new TupleBatch(entry.getKey(), rows); + if (isFinal && result.getEndRow() == rowCount) { + result.setTerminationFlag(true); + } if (forwardOnly) { batches.remove(entry.getKey()); } } - result.setDataTypes(types); if (isFinal && result.getEndRow() == rowCount) { result.setTerminationFlag(true); } @@ -246,8 +236,8 @@ if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.DETAIL)) { LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, "Removing TupleBuffer:", this.tupleSourceID); //$NON-NLS-1$ } - if (this.lobStore != null) { - this.lobStore.remove(); + if (this.lobManager != null) { + this.lobManager.remove(); } this.batchBuffer = null; purge(); @@ -365,18 +355,11 @@ } public void setPrefersMemory(boolean prefersMemory) { - this.prefersMemory = prefersMemory; - for (ManagedBatch batch : this.batches.values()) { - batch.setPrefersMemory(prefersMemory); - } + this.manager.setPrefersMemory(prefersMemory); } - public boolean isPrefersMemory() { - return prefersMemory; - } - public String[] getTypes() { - return types; + return manager.getTypes(); } public int getLobCount() { Added: trunk/engine/src/main/java/org/teiid/common/buffer/impl/BitSetTree.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/impl/BitSetTree.java (rev 0) +++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/BitSetTree.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -0,0 +1,106 @@ +/* + * JBoss, Home of Professional Open Source. + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301 USA. + */ + + package org.teiid.common.buffer.impl; + +import java.util.BitSet; + +/** + * Extends a {@link BitSet} by adding a cumulative total and a + * first level index to speed queries against large bitsets. + */ +public class BitSetTree { + + public static final int MAX_INDEX = (1 << 24) - 1; + + private int bitsSet; + private int totalBits; + private short[] topVals = new short[1 << 9]; + private BitSet bitSet = new BitSet(); + + public boolean get(int index) { + if (index > MAX_INDEX) { + throw new ArrayIndexOutOfBoundsException(index); + } + return bitSet.get(index); + } + + /** + * Set the given bit at the index. It's expected that the + * bit currently has the opposite value. + * @param bitIndex + * @param value + */ + public void set(int bitIndex, boolean value) { + if (bitIndex > MAX_INDEX) { + throw new ArrayIndexOutOfBoundsException(bitIndex); + } + bitSet.set(bitIndex, value); + if (bitIndex >= totalBits) { + totalBits = bitIndex + 1; + } + int topIndex = bitIndex >>> 15; + int increment = value?1:-1; + bitsSet+=increment; + topVals[topIndex]+=increment; + } + + public int getTotalBits() { + return totalBits; + } + + public int getBitsSet() { + return bitsSet; + } + + public int nextClearBit(int fromIndex) { + int start = fromIndex >> 15; + for (int i = start; i < topVals.length; i++) { + if (topVals[i] < Short.MAX_VALUE) { + int searchFrom = fromIndex; + if (i > start) { + searchFrom = i << 15; + } + return bitSet.nextClearBit(searchFrom); + } + } + return -1; + } + + public int nextSetBit(int fromIndex) { + if (bitsSet == 0) { + return -1; + } + int start = fromIndex >> 15; + for (int i = fromIndex >> 15; i < topVals.length; i++) { + if (topVals[i] > 0) { + int searchFrom = fromIndex; + if (i > start) { + searchFrom = i << 15; + } + return bitSet.nextSetBit(searchFrom); + } + } + return -1; + } + +} Property changes on: trunk/engine/src/main/java/org/teiid/common/buffer/impl/BitSetTree.java ___________________________________________________________________ Added: svn:mime-type + text/plain Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -22,7 +22,6 @@ package org.teiid.common.buffer.impl; -import java.io.BufferedInputStream; import java.io.IOException; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; @@ -31,41 +30,41 @@ import java.lang.ref.SoftReference; import java.lang.ref.WeakReference; import java.util.Arrays; -import java.util.Comparator; -import java.util.HashMap; +import java.util.Collection; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.TreeMap; -import java.util.TreeSet; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.Condition; -import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantLock; -import java.util.concurrent.locks.ReentrantReadWriteLock; +import org.teiid.client.BatchSerializer; +import org.teiid.common.buffer.AutoCleanupUtil; import org.teiid.common.buffer.BatchManager; import org.teiid.common.buffer.BufferManager; +import org.teiid.common.buffer.Cache; +import org.teiid.common.buffer.CacheEntry; import org.teiid.common.buffer.FileStore; import org.teiid.common.buffer.LobManager; import org.teiid.common.buffer.STree; +import org.teiid.common.buffer.Serializer; import org.teiid.common.buffer.StorageManager; -import org.teiid.common.buffer.TupleBatch; import org.teiid.common.buffer.TupleBuffer; -import org.teiid.common.buffer.BatchManager.ManagedBatch; +import org.teiid.common.buffer.AutoCleanupUtil.Removable; +import org.teiid.common.buffer.LobManager.ReferenceMode; import org.teiid.core.TeiidComponentException; import org.teiid.core.TeiidRuntimeException; import org.teiid.core.types.DataTypeManager; -import org.teiid.core.util.Assertion; +import org.teiid.core.types.DataTypeManager.WeakReferenceHashedValueCache; import org.teiid.dqp.internal.process.DQPConfiguration; import org.teiid.logging.LogConstants; import org.teiid.logging.LogManager; import org.teiid.logging.MessageLevel; -import org.teiid.query.QueryPlugin; import org.teiid.query.processor.relational.ListNestedSortComparator; import org.teiid.query.sql.symbol.Expression; @@ -74,477 +73,201 @@ * <p>Default implementation of BufferManager.</p> * Responsible for creating/tracking TupleBuffers and providing access to the StorageManager. * </p> - * The buffering strategy attempts to purge batches from the least recently used TupleBuffer - * from before (which wraps around circularly) the last used batch. This attempts to compensate - * for our tendency to read buffers in a forward manner. If our processing algorithms are changed - * to use alternating ascending/descending access, then the buffering approach could be replaced - * with a simple LRU. * - * TODO: allow for cached stores to use lru - (result set/mat view) - * TODO: account for row/content based sizing (difficult given value sharing) * TODO: add detection of pinned batches to prevent unnecessary purging of non-persistent batches * - this is not necessary for already persistent batches, since we hold a weak reference */ public class BufferManagerImpl implements BufferManager, StorageManager { - private static final int TARGET_BYTES_PER_ROW = 1 << 11; //2k bytes per row - private static final int IO_BUFFER_SIZE = 1 << 14; - private static final int COMPACTION_THRESHOLD = 1 << 25; //start checking at 32 megs - - private final class CleanupHook implements org.teiid.common.buffer.BatchManager.CleanupHook { - - private Long id; - private WeakReference<BatchManagerImpl> ref; - - CleanupHook(Long id, WeakReference<BatchManagerImpl> batchManager) { - this.id = id; - this.ref = batchManager; - } - - public void cleanup() { - BatchManagerImpl batchManager = ref.get(); - if (batchManager == null) { - return; - } - cleanupManagedBatch(batchManager, id); - } - - } - - private final class BatchManagerImpl implements BatchManager { - final String id; - volatile FileStore store; - Map<Long, long[]> physicalMapping = new HashMap<Long, long[]>(); - long tail; - ConcurrentSkipListSet<Long> freed = new ConcurrentSkipListSet<Long>(); - ReadWriteLock compactionLock = new ReentrantReadWriteLock(); - AtomicLong unusedSpace = new AtomicLong(); - private int[] lobIndexes; + private final class BatchManagerImpl implements BatchManager, Serializer<List<? extends List<?>>> { + final Long id; SizeUtility sizeUtility; private WeakReference<BatchManagerImpl> ref = new WeakReference<BatchManagerImpl>(this); + AtomicBoolean prefersMemory = new AtomicBoolean(); + String[] types; + private LobManager lobManager; - private BatchManagerImpl(String newID, int[] lobIndexes) { + private BatchManagerImpl(Long newID, String[] types) { this.id = newID; - this.store = createFileStore(id); - this.store.setCleanupReference(this); - this.lobIndexes = lobIndexes; - this.sizeUtility = new SizeUtility(); + this.sizeUtility = new SizeUtility(types); + this.types = types; + cache.createCacheGroup(newID); } - private void freeBatch(Long batch) { - long[] info = physicalMapping.remove(batch); - if (info != null) { - unusedSpace.addAndGet(info[1]); - if (info[0] + info[1] == tail) { - tail -= info[1]; - } - } + @Override + public Long getId() { + return id; } - public FileStore createStorage(String prefix) { - return createFileStore(id+prefix); + public void setLobManager(LobManager lobManager) { + this.lobManager = lobManager; } - + @Override - public ManagedBatch createManagedBatch(TupleBatch batch, boolean softCache) - throws TeiidComponentException { - ManagedBatchImpl mbi = new ManagedBatchImpl(batch, ref, softCache); - mbi.addToCache(false); - persistBatchReferences(); - return mbi; + public String[] getTypes() { + return types; } - private long getOffset() throws TeiidComponentException { - if (store.getLength() <= compactionThreshold || unusedSpace.get() * 4 <= store.getLength() * 3) { - return tail; - } - if (LogManager.isMessageToBeRecorded(LogConstants.CTX_DQP, MessageLevel.DETAIL)) { - LogManager.logDetail(LogConstants.CTX_DQP, "Running full compaction on", id); //$NON-NLS-1$ - } - byte[] buffer = new byte[IO_BUFFER_SIZE]; - TreeSet<long[]> bySize = new TreeSet<long[]>(new Comparator<long[]>() { - @Override - public int compare(long[] o1, long[] o2) { - int signum = Long.signum(o1[1] - o2[1]); - if (signum == 0) { - //take the upper address first - return Long.signum(o2[0] - o1[0]); - } - return signum; - } - }); - TreeSet<long[]> byAddress = new TreeSet<long[]>(new Comparator<long[]>() { - - @Override - public int compare(long[] o1, long[] o2) { - return Long.signum(o1[0] - o2[0]); - } - }); - bySize.addAll(physicalMapping.values()); - byAddress.addAll(physicalMapping.values()); - long lastEndAddress = 0; - unusedSpace.set(0); - long minFreeSpace = 1 << 11; - while (!byAddress.isEmpty()) { - long[] info = byAddress.pollFirst(); - bySize.remove(info); - - long currentOffset = info[0]; - long space = currentOffset - lastEndAddress; - while (space > 0 && !bySize.isEmpty()) { - long[] smallest = bySize.first(); - if (smallest[1] > space) { - break; - } - bySize.pollFirst(); - byAddress.remove(smallest); - move(smallest, lastEndAddress, buffer); - space -= smallest[1]; - lastEndAddress += smallest[1]; - } - - if (space <= minFreeSpace) { - unusedSpace.addAndGet(space); - } else { - move(info, lastEndAddress, buffer); - } - lastEndAddress = info[0] + info[1]; - } - long oldLength = store.getLength(); - store.truncate(lastEndAddress); - tail = lastEndAddress; - if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.DETAIL)) { - LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, "Compacted store", id, "pre-size", oldLength, "post-size", store.getLength()); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ - } - return tail; + @Override + public boolean prefersMemory() { + return prefersMemory.get(); } - private void move(long[] toMove, long newOffset, byte[] buffer) throws TeiidComponentException { - long oldOffset = toMove[0]; - toMove[0] = newOffset; - int size = (int)toMove[1]; - while (size > 0) { - int toWrite = Math.min(IO_BUFFER_SIZE, size); - store.readFully(oldOffset, buffer, 0, toWrite); - store.write(newOffset, buffer, 0, toWrite); - size -= toWrite; - oldOffset += toWrite; - newOffset += toWrite; - } + @Override + public void setPrefersMemory(boolean prefers) { + //TODO: it's only expected to move from not preferring to prefefring + this.prefersMemory.set(prefers); } - + @Override - public void remove() { - this.store.remove(); + public boolean useSoftCache() { + return prefersMemory.get(); } @Override - public String toString() { - return id; + public Reference<? extends BatchManager> getBatchManagerReference() { + return ref; } - } - - /** - * Holder for active batches - */ - private class TupleBufferInfo { - TreeMap<Long, ManagedBatchImpl> batches = new TreeMap<Long, ManagedBatchImpl>(); - Long lastUsed = null; - ManagedBatchImpl removeBatch(Long row) { - ManagedBatchImpl result = batches.remove(row); - if (result != null) { - activeBatchKB -= result.sizeEstimate; - if (result.softCache) { - BatchSoftReference ref = (BatchSoftReference)result.batchReference; - if (ref != null) { - maxReserveKB += ref.sizeEstimate; - ref.sizeEstimate = 0; - ref.clear(); + @Override + public Long createManagedBatch(List<? extends List<?>> batch) + throws TeiidComponentException { + int sizeEstimate = getSizeEstimate(batch); + Long oid = batchAdded.getAndIncrement(); + CacheEntry ce = new CacheEntry(oid); + ce.setObject(batch); + ce.setSizeEstimate(sizeEstimate); + ce.setSerializer(this.ref); + return addCacheEntry(ce, this); + } + + @Override + public List<? extends List<?>> deserialize(ObjectInputStream ois) + throws IOException, ClassNotFoundException { + List<? extends List<?>> batch = BatchSerializer.readBatch(ois, types); + if (lobManager != null) { + for (List<?> list : batch) { + try { + lobManager.updateReferences(list, ReferenceMode.ATTACH); + } catch (TeiidComponentException e) { + throw new TeiidRuntimeException(e); } } } - return result; + return batch; } - } - - private final class ManagedBatchImpl implements ManagedBatch { - private boolean persistent; - private boolean softCache; - private volatile TupleBatch activeBatch; - private volatile Reference<TupleBatch> batchReference; - private WeakReference<BatchManagerImpl> managerRef; - private Long id; - private LobManager lobManager; - private int sizeEstimate; - public ManagedBatchImpl(TupleBatch batch, WeakReference<BatchManagerImpl> ref, boolean softCache) { - this.softCache = softCache; - id = batchAdded.incrementAndGet(); - this.activeBatch = batch; - this.managerRef = ref; - BatchManagerImpl batchManager = ref.get(); - if (batchManager.lobIndexes != null) { - this.lobManager = new LobManager(); - } - sizeEstimate = (int) Math.max(1, batchManager.sizeUtility.getBatchSize(batch) / 1024); - if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.TRACE)) { - LogManager.logTrace(LogConstants.CTX_BUFFER_MGR, "Add batch to BufferManager", id, "with size estimate", sizeEstimate); //$NON-NLS-1$ //$NON-NLS-2$ - } + @Override + public void serialize(List<? extends List<?>> obj, + ObjectOutputStream oos) throws IOException { + //it's expected that the containing structure has updated the lob manager + BatchSerializer.writeBatch(oos, types, obj); } - @Override - public void setPrefersMemory(boolean prefers) { - this.softCache = prefers; - //TODO: could recreate the reference + public int getSizeEstimate(List<? extends List<?>> obj) { + return (int) Math.max(1, sizeUtility.getBatchSize(obj) / 1024); } - - private void addToCache(boolean update) { - BatchManagerImpl batchManager = managerRef.get(); - if (batchManager == null) { - remove(); - return; - } - synchronized (activeBatches) { - TupleBatch batch = this.activeBatch; - if (batch == null) { - return; //already removed - } - activeBatchKB += sizeEstimate; - TupleBufferInfo tbi = null; - if (update) { - tbi = activeBatches.remove(batchManager.id); - } else { - tbi = activeBatches.get(batchManager.id); - } - if (tbi == null) { - tbi = new TupleBufferInfo(); - update = true; - } - if (update) { - activeBatches.put(batchManager.id, tbi); - } - tbi.batches.put(this.id, this); - } - } - + + @SuppressWarnings("unchecked") @Override - public TupleBatch getBatch(boolean cache, String[] types) throws TeiidComponentException { - BatchManagerImpl batchManager = managerRef.get(); - if (batchManager == null) { - remove(); - throw new AssertionError("Already removed"); //$NON-NLS-1$ - } + public List<List<?>> getBatch(Long batch, boolean retain) + throws TeiidComponentException { + cleanSoftReferences(); long reads = readAttempts.incrementAndGet(); if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.TRACE)) { - LogManager.logTrace(LogConstants.CTX_BUFFER_MGR, batchManager.id, "getting batch", reads, "reference hits", referenceHit.get()); //$NON-NLS-1$ //$NON-NLS-2$ + LogManager.logTrace(LogConstants.CTX_BUFFER_MGR, id, "getting batch", batch, "total reads", reads, "reference hits", referenceHit.get()); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ } - synchronized (activeBatches) { - for (int i = 0; i < 10; i++) { - BatchSoftReference ref = (BatchSoftReference)SOFT_QUEUE.poll(); - if (ref == null) { - break; - } - maxReserveKB += ref.sizeEstimate; - ref.sizeEstimate = 0; - ref.clear(); - } - TupleBufferInfo tbi = activeBatches.remove(batchManager.id); - if (tbi != null) { - boolean put = true; - if (!cache) { - tbi.removeBatch(this.id); - if (tbi.batches.isEmpty()) { - put = false; - } - } - if (put) { - tbi.lastUsed = this.id; - activeBatches.put(batchManager.id, tbi); - } - } + CacheEntry ce = fastGet(batch, prefersMemory.get(), retain); + if (ce != null) { + return (List<List<?>>)(!retain?ce.nullOut():ce.getObject()); } - persistBatchReferences(); synchronized (this) { - TupleBatch batch = this.activeBatch; - if (batch != null){ - return batch; + ce = fastGet(batch, prefersMemory.get(), retain); + if (ce != null) { + return (List<List<?>>)(!retain?ce.nullOut():ce.getObject()); } - Reference<TupleBatch> ref = this.batchReference; - this.batchReference = null; - if (ref != null) { - batch = ref.get(); - if (batch != null) { - if (cache) { - this.activeBatch = batch; - addToCache(true); - } - referenceHit.getAndIncrement(); - return batch; - } - } long count = readCount.incrementAndGet(); if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.TRACE)) { - LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, batchManager.id, id, "reading batch from disk, total reads:", count); //$NON-NLS-1$ + LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, id, id, "reading batch", batch, "from storage, total reads:", count); //$NON-NLS-1$ //$NON-NLS-2$ } - try { - batchManager.compactionLock.readLock().lock(); - long[] info = batchManager.physicalMapping.get(this.id); - ObjectInputStream ois = new ObjectInputStream(new BufferedInputStream(batchManager.store.createInputStream(info[0]), IO_BUFFER_SIZE)); - batch = new TupleBatch(); - batch.setRowOffset(ois.readInt()); - batch.setDataTypes(types); - batch.readExternal(ois); - batch.setDataTypes(null); - if (lobManager != null) { - for (List<?> tuple : batch.getTuples()) { - lobManager.updateReferences(batchManager.lobIndexes, tuple); - } - } - if (cache) { - this.activeBatch = batch; - addToCache(true); - } - return batch; - } catch(IOException e) { - throw new TeiidComponentException(e, QueryPlugin.Util.getString("FileStoreageManager.error_reading", batchManager.id)); //$NON-NLS-1$ - } catch (ClassNotFoundException e) { - throw new TeiidComponentException(e, QueryPlugin.Util.getString("FileStoreageManager.error_reading", batchManager.id)); //$NON-NLS-1$ - } finally { - batchManager.compactionLock.readLock().unlock(); - } - } - } - - public synchronized void persist() throws TeiidComponentException { - final BatchManagerImpl batchManager = managerRef.get(); - if (batchManager == null) { - remove(); - return; - } - boolean lockheld = false; - try { - TupleBatch batch = activeBatch; - if (batch != null) { - if (!persistent) { - long count = writeCount.incrementAndGet(); - if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.DETAIL)) { - LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, batchManager.id, id, "writing batch to disk, total writes: ", count); //$NON-NLS-1$ - } - if (lobManager != null) { - for (List<?> tuple : batch.getTuples()) { - lobManager.updateReferences(batchManager.lobIndexes, tuple); - } - } - batchManager.compactionLock.writeLock().lock(); - Long free = null; - while ((free = batchManager.freed.pollFirst()) != null) { - batchManager.freeBatch(free); - } - lockheld = true; - final long offset = batchManager.getOffset(); - ExtensibleBufferedOutputStream fsos = new ExtensibleBufferedOutputStream(new byte[IO_BUFFER_SIZE]) { - - @Override - protected void flushDirect() throws IOException { - try { - batchManager.store.write(offset + bytesWritten, buf, 0, count); - } catch (TeiidComponentException e) { - throw new IOException(e); - } - } - }; - ObjectOutputStream oos = new ObjectOutputStream(fsos); - oos.writeInt(batch.getBeginRow()); - batch.writeExternal(oos); - oos.close(); - long size = fsos.getBytesWritten(); - long[] info = new long[] {offset, size}; - batchManager.physicalMapping.put(this.id, info); - batchManager.tail = Math.max(batchManager.tail, offset + size); - if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.TRACE)) { - LogManager.logTrace(LogConstants.CTX_BUFFER_MGR, batchManager.id, id, "batch written starting at:", offset); //$NON-NLS-1$ - } - } - if (softCache) { - this.batchReference = new BatchSoftReference(batch, SOFT_QUEUE, sizeEstimate); - synchronized (activeBatches) { - maxReserveKB -= sizeEstimate; - } - } else if (useWeakReferences) { - this.batchReference = new WeakReference<TupleBatch>(batch); - } + ce = cache.get(batch, this); + if (ce == null) { + throw new AssertionError("Batch not found in storage " + batch); //$NON-NLS-1$ } - } catch (IOException e) { - throw new TeiidComponentException(e); - } catch (Throwable e) { - throw new TeiidComponentException(e); - } finally { - persistent = true; - activeBatch = null; - if (lockheld) { - batchManager.compactionLock.writeLock().unlock(); + if (!retain) { + cache.remove(this.id, batch); } - } + ce.setSerializer(this.ref); + if (retain) { + addMemoryEntry(ce); + } + } + return (List<List<?>>)ce.getObject(); } + + @Override + public void remove(Long batch) { + cleanSoftReferences(); + BufferManagerImpl.this.remove(id, batch, prefersMemory.get()); + } + @Override public void remove() { - activeBatch = null; - batchReference = null; - BatchManagerImpl batchManager = managerRef.get(); - if (batchManager != null) { - cleanupManagedBatch(batchManager, id); - } + removeCacheGroup(id, prefersMemory.get()); } - + @Override - public CleanupHook getCleanupHook() { - return new CleanupHook(id, managerRef); - } - - @Override public String toString() { - return "ManagedBatch " + managerRef.get() + " " + this.id + " " + activeBatch; //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ + return id.toString(); } } - private static class BatchSoftReference extends SoftReference<TupleBatch> { + private static class BatchSoftReference extends SoftReference<CacheEntry> { private int sizeEstimate; + private Long key; - public BatchSoftReference(TupleBatch referent, - ReferenceQueue<? super TupleBatch> q, int sizeEstimate) { + public BatchSoftReference(CacheEntry referent, + ReferenceQueue<? super CacheEntry> q, int sizeEstimate) { super(referent, q); this.sizeEstimate = sizeEstimate; + this.key = referent.getId(); } } + + private static final int TARGET_BYTES_PER_ROW = 1 << 11; //2k bytes per row + private static ReferenceQueue<CacheEntry> SOFT_QUEUE = new ReferenceQueue<CacheEntry>(); - private static ReferenceQueue<? super TupleBatch> SOFT_QUEUE = new ReferenceQueue<TupleBatch>(); - // Configuration private int connectorBatchSize = BufferManager.DEFAULT_CONNECTOR_BATCH_SIZE; private int processorBatchSize = BufferManager.DEFAULT_PROCESSOR_BATCH_SIZE; //set to acceptable defaults for testing private int maxProcessingKB = 1 << 11; private Integer maxProcessingKBOrig; - private int maxReserveKB = 1 << 25; + private AtomicInteger maxReserveKB = new AtomicInteger(1 << 18); private volatile int reserveBatchKB; private int maxActivePlans = DQPConfiguration.DEFAULT_MAX_ACTIVE_PLANS; //used as a hint to set the reserveBatchKB private boolean useWeakReferences = true; private boolean inlineLobs = true; private int targetBytesPerRow = TARGET_BYTES_PER_ROW; - private int compactionThreshold = COMPACTION_THRESHOLD; private ReentrantLock lock = new ReentrantLock(true); private Condition batchesFreed = lock.newCondition(); - private volatile int activeBatchKB = 0; - private Map<String, TupleBufferInfo> activeBatches = new LinkedHashMap<String, TupleBufferInfo>(); + private AtomicInteger activeBatchKB = new AtomicInteger(); + + //tiered memory entries. the first tier is just a queue of adds/gets. once accessed again, the entry moves to the tenured tier. + private LinkedHashMap<Long, CacheEntry> memoryEntries = new LinkedHashMap<Long, CacheEntry>(16, .75f, false); + private LinkedHashMap<Long, CacheEntry> tenuredMemoryEntries = new LinkedHashMap<Long, CacheEntry>(16, .75f, true); + + private WeakReferenceHashedValueCache<CacheEntry> weakReferenceCache; //limited size based upon the memory settings + private ConcurrentHashMap<Long, BatchSoftReference> softCache = new ConcurrentHashMap<Long, BatchSoftReference>(); //"unlimitted" size maintained by reference queue + + private Cache cache; + private Map<String, TupleReference> tupleBufferMap = new ConcurrentHashMap<String, TupleReference>(); private ReferenceQueue<TupleBuffer> tupleBufferQueue = new ReferenceQueue<TupleBuffer>(); - private StorageManager diskMgr; - private AtomicLong tsId = new AtomicLong(); private AtomicLong batchAdded = new AtomicLong(); private AtomicLong readCount = new AtomicLong(); @@ -603,60 +326,42 @@ this.processorBatchSize = processorBatchSize; } - /** - * Add a storage manager to this buffer manager, order is unimportant - * @param storageManager Storage manager to add - */ - public void setStorageManager(StorageManager storageManager) { - Assertion.isNotNull(storageManager); - Assertion.isNull(diskMgr); - this.diskMgr = storageManager; - } - - public StorageManager getStorageManager() { - return diskMgr; - } - @Override public TupleBuffer createTupleBuffer(final List elements, String groupName, TupleSourceType tupleSourceType) { - final String newID = String.valueOf(this.tsId.getAndIncrement()); + final Long newID = this.tsId.getAndIncrement(); int[] lobIndexes = LobManager.getLobIndexes(elements); - BatchManager batchManager = new BatchManagerImpl(newID, lobIndexes); - TupleBuffer tupleBuffer = new TupleBuffer(batchManager, newID, elements, lobIndexes, getProcessorBatchSize(elements)); + String[] types = TupleBuffer.getTypeNames(elements); + BatchManagerImpl batchManager = createBatchManager(newID, types); + LobManager lobManager = null; + FileStore lobStore = null; + if (lobIndexes != null) { + lobStore = createFileStore(newID + "_lobs"); //$NON-NLS-1$ + lobManager = new LobManager(lobIndexes, lobStore); + batchManager.setLobManager(lobManager); + } + TupleBuffer tupleBuffer = new TupleBuffer(batchManager, String.valueOf(newID), elements, lobManager, getProcessorBatchSize(elements)); + if (lobStore != null) { + AutoCleanupUtil.setCleanupReference(batchManager, lobStore); + } if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.DETAIL)) { - LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, "Creating TupleBuffer:", newID, elements, Arrays.toString(tupleBuffer.getTypes()), "of type", tupleSourceType); //$NON-NLS-1$ //$NON-NLS-2$ + LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, "Creating TupleBuffer:", newID, elements, Arrays.toString(types), "of type", tupleSourceType); //$NON-NLS-1$ //$NON-NLS-2$ } tupleBuffer.setInlineLobs(inlineLobs); return tupleBuffer; } - private void cleanupManagedBatch(BatchManagerImpl batchManager, Long id) { - synchronized (activeBatches) { - TupleBufferInfo tbi = activeBatches.get(batchManager.id); - if (tbi != null && tbi.removeBatch(id) != null) { - if (tbi.batches.isEmpty()) { - activeBatches.remove(batchManager.id); - } - } - } - - if (batchManager.compactionLock.writeLock().tryLock()) { - try { - batchManager.freeBatch(id); - } finally { - batchManager.compactionLock.writeLock().unlock(); - } - } else { - batchManager.freed.add(id); - } - } - public STree createSTree(final List elements, String groupName, int keyLength) { - String newID = String.valueOf(this.tsId.getAndIncrement()); + Long newID = this.tsId.getAndIncrement(); int[] lobIndexes = LobManager.getLobIndexes(elements); - BatchManager bm = new BatchManagerImpl(newID, lobIndexes); - BatchManager keyManager = new BatchManagerImpl(String.valueOf(this.tsId.getAndIncrement()), null); + String[] types = TupleBuffer.getTypeNames(elements); + BatchManagerImpl bm = createBatchManager(newID, types); + LobManager lobManager = null; + if (lobIndexes != null) { + lobManager = new LobManager(lobIndexes, null); //persistence is not expected yet - later we might utilize storage for out-of-line lob values + bm.setLobManager(lobManager); + } + BatchManager keyManager = createBatchManager(this.tsId.getAndIncrement(), Arrays.copyOf(types, keyLength)); int[] compareIndexes = new int[keyLength]; for (int i = 1; i < compareIndexes.length; i++) { compareIndexes[i] = i; @@ -664,16 +369,33 @@ if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.DETAIL)) { LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, "Creating STree:", newID); //$NON-NLS-1$ } - return new STree(keyManager, bm, new ListNestedSortComparator(compareIndexes), getProcessorBatchSize(elements), getProcessorBatchSize(elements.subList(0, keyLength)), keyLength, TupleBuffer.getTypeNames(elements)); + return new STree(keyManager, bm, new ListNestedSortComparator(compareIndexes), getProcessorBatchSize(elements), getProcessorBatchSize(elements.subList(0, keyLength)), keyLength, lobManager); } + private BatchManagerImpl createBatchManager(final Long newID, String[] types) { + BatchManagerImpl bm = new BatchManagerImpl(newID, types); + final AtomicBoolean prefersMemory = bm.prefersMemory; + AutoCleanupUtil.setCleanupReference(bm, new Removable() { + + @Override + public void remove() { + BufferManagerImpl.this.removeCacheGroup(newID, prefersMemory.get()); + } + }); + return bm; + } + @Override public FileStore createFileStore(String name) { if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.DETAIL)) { LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, "Creating FileStore:", name); //$NON-NLS-1$ } - return this.diskMgr.createFileStore(name); + return this.cache.createFileStore(name); } + + public Cache getCache() { + return cache; + } public void setMaxActivePlans(int maxActivePlans) { this.maxActivePlans = maxActivePlans; @@ -684,23 +406,23 @@ } public void setMaxReserveKB(int maxReserveBatchKB) { - this.maxReserveKB = maxReserveBatchKB; + this.maxReserveKB.set(maxReserveBatchKB); } @Override public void initialize() throws TeiidComponentException { int maxMemory = (int)Math.min(Runtime.getRuntime().maxMemory() / 1024, Integer.MAX_VALUE); maxMemory -= 300 * 1024; //assume 300 megs of overhead for the AS/system stuff - if (maxReserveKB < 0) { - this.maxReserveKB = 0; + if (getMaxReserveKB() < 0) { + this.setMaxReserveKB(0); int one_gig = 1024 * 1024; if (maxMemory > one_gig) { //assume 75% of the memory over the first gig - this.maxReserveKB += (int)Math.max(0, (maxMemory - one_gig) * .75); + this.maxReserveKB.addAndGet(((int)Math.max(0, (maxMemory - one_gig) * .75))); } - this.maxReserveKB += Math.max(0, Math.min(one_gig, maxMemory) * .5); + this.maxReserveKB.addAndGet(((int)Math.max(0, Math.min(one_gig, maxMemory) * .5))); } - this.reserveBatchKB = this.maxReserveKB; + this.reserveBatchKB = this.getMaxReserveKB(); if (this.maxProcessingKBOrig == null) { //store the config value so that we can be reinitialized (this is not a clean approach) this.maxProcessingKBOrig = this.maxProcessingKB; @@ -708,6 +430,11 @@ if (this.maxProcessingKBOrig < 0) { this.maxProcessingKB = Math.max(Math.min(8 * processorBatchSize, Integer.MAX_VALUE), (int)(.1 * maxMemory)/maxActivePlans); } + if (useWeakReferences) { + int memoryBatches = (this.maxProcessingKB * maxActivePlans + this.getMaxReserveKB()) / (processorBatchSize * targetBytesPerRow / 1024); + int logSize = 39 - Integer.numberOfLeadingZeros(memoryBatches); + weakReferenceCache = new WeakReferenceHashedValueCache<CacheEntry>(Math.min(20, logSize)); + } } @Override @@ -736,7 +463,7 @@ try { if (mode == BufferReserveMode.WAIT) { //don't wait for more than is available - int waitCount = Math.min(count, this.maxReserveKB); + int waitCount = Math.min(count, this.getMaxReserveKB()); while (waitCount > 0 && waitCount > this.reserveBatchKB) { try { batchesFreed.await(100, TimeUnit.MILLISECONDS); @@ -760,46 +487,186 @@ } void persistBatchReferences() { - if (activeBatchKB == 0 || activeBatchKB <= reserveBatchKB) { - int memoryCount = activeBatchKB + maxReserveKB - reserveBatchKB; + if (activeBatchKB.get() == 0 || activeBatchKB.get() <= reserveBatchKB) { + int memoryCount = activeBatchKB.get() + getMaxReserveKB() - reserveBatchKB; if (DataTypeManager.isValueCacheEnabled()) { - if (memoryCount < maxReserveKB / 8) { + if (memoryCount < getMaxReserveKB() / 8) { DataTypeManager.setValueCacheEnabled(false); } - } else if (memoryCount > maxReserveKB / 4) { + } else if (memoryCount > getMaxReserveKB() / 4) { DataTypeManager.setValueCacheEnabled(true); } return; } + boolean first = true; while (true) { - ManagedBatchImpl mb = null; - synchronized (activeBatches) { - if (activeBatchKB == 0 || activeBatchKB < reserveBatchKB * .8) { + CacheEntry ce = null; + synchronized (memoryEntries) { + if (activeBatchKB.get() == 0 || activeBatchKB.get() < reserveBatchKB * .8) { break; } - Iterator<TupleBufferInfo> iter = activeBatches.values().iterator(); - TupleBufferInfo tbi = iter.next(); - Map.Entry<Long, ManagedBatchImpl> entry = null; - if (tbi.lastUsed != null) { - entry = tbi.batches.floorEntry(tbi.lastUsed - 1); + if (first) { //let one entry per persist cycle loose its tenure. this helps us be more write avoident. + first = false; + if (!tenuredMemoryEntries.isEmpty()) { + Iterator<Map.Entry<Long, CacheEntry>> iter = tenuredMemoryEntries.entrySet().iterator(); + Map.Entry<Long, CacheEntry> entry = iter.next(); + iter.remove(); + memoryEntries.put(entry.getKey(), entry.getValue()); + } } - if (entry == null) { - entry = tbi.batches.lastEntry(); + LinkedHashMap<Long, CacheEntry> toDrain = memoryEntries; + if (memoryEntries.isEmpty()) { + toDrain = tenuredMemoryEntries; + if (tenuredMemoryEntries.isEmpty()) { + break; + } + } + Iterator<CacheEntry> iter = toDrain.values().iterator(); + ce = iter.next(); + iter.remove(); + activeBatchKB.addAndGet(-ce.getSizeEstimate()); + } + persist(ce); + } + } + + void persist(CacheEntry ce) { + Serializer<?> s = ce.getSerializer().get(); + if (s == null) { + return; + } + if (!ce.isPersistent()) { + long count = writeCount.incrementAndGet(); + if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.DETAIL)) { + LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, ce.getId(), "writing batch to storage, total writes: ", count); //$NON-NLS-1$ + } + cache.add(ce, s); + ce.setPersistent(true); + } + if (s.useSoftCache()) { + createSoftReference(ce); + } else if (useWeakReferences) { + weakReferenceCache.getValue(ce); //a get will set the value + } + } + + private void createSoftReference(CacheEntry ce) { + BatchSoftReference ref = new BatchSoftReference(ce, SOFT_QUEUE, ce.getSizeEstimate()); + softCache.put(ce.getId(), ref); + maxReserveKB.addAndGet(- ce.getSizeEstimate()/2); + } + + /** + * Get a CacheEntry without hitting the cache + */ + CacheEntry fastGet(Long batch, boolean prefersMemory, boolean retain) { + CacheEntry ce = null; + synchronized (memoryEntries) { + if (retain) { + ce = tenuredMemoryEntries.get(batch); + if (ce == null) { + ce = memoryEntries.remove(batch); + if (ce != null) { + tenuredMemoryEntries.put(batch, ce); + } } - tbi.removeBatch(entry.getKey()); - if (tbi.batches.isEmpty()) { - iter.remove(); + } else { + ce = tenuredMemoryEntries.remove(batch); + if (ce == null) { + ce = memoryEntries.remove(batch); } - mb = entry.getValue(); } - try { - mb.persist(); - } catch (TeiidComponentException e) { - LogManager.logError(LogConstants.CTX_BUFFER_MGR, e, "Error persisting batch, attempts to read that batch later will result in an exception"); //$NON-NLS-1$ + } + if (ce != null) { + if (!retain) { + BufferManagerImpl.this.remove(ce, true); } + return ce; } + if (prefersMemory) { + BatchSoftReference bsr = softCache.get(batch); + if (bsr != null) { + ce = bsr.get(); + if (ce != null) { + softCache.remove(batch); + maxReserveKB.addAndGet(bsr.sizeEstimate/2); + } + } + } else if (useWeakReferences) { + ce = weakReferenceCache.getByHash(batch); + if (ce == null || !ce.getId().equals(batch)) { + return null; + } + } + if (ce != null && ce.getObject() != null) { + referenceHit.getAndIncrement(); + if (retain) { + addMemoryEntry(ce); + } else { + BufferManagerImpl.this.remove(ce, false); + } + return ce; + } + return null; } + void remove(Long gid, Long batch, boolean prefersMemory) { + CacheEntry ce = fastGet(batch, prefersMemory, false); + if (ce == null) { + cache.remove(gid, batch); + } else { + ce.nullOut(); + } + } + + private void remove(CacheEntry ce, boolean inMemory) { + if (inMemory) { + activeBatchKB.addAndGet(-ce.getSizeEstimate()); + } + Serializer<?> s = ce.getSerializer().get(); + if (s != null) { + cache.remove(s.getId(), ce.getId()); + } + } + + Long addCacheEntry(CacheEntry ce, Serializer<?> s) { + if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.TRACE)) { + LogManager.logTrace(LogConstants.CTX_BUFFER_MGR, "Add batch to BufferManager", ce.getId(), "with size estimate", ce.getSizeEstimate()); //$NON-NLS-1$ //$NON-NLS-2$ + } + cache.addToCacheGroup(s.getId(), ce.getId()); + addMemoryEntry(ce); + return ce.getId(); + } + + void addMemoryEntry(CacheEntry ce) { + persistBatchReferences(); + synchronized (memoryEntries) { + memoryEntries.put(ce.getId(), ce); + } + activeBatchKB.getAndAdd(ce.getSizeEstimate()); + } + + void removeCacheGroup(Long id, boolean prefersMemory) { + cleanSoftReferences(); + Collection<Long> vals = cache.removeCacheGroup(id); + for (Long val : vals) { + fastGet(val, prefersMemory, false); + } + } + + void cleanSoftReferences() { + for (int i = 0; i < 10; i++) { + BatchSoftReference ref = (BatchSoftReference)SOFT_QUEUE.poll(); + if (ref == null) { + break; + } + softCache.remove(ref.key); + maxReserveKB.addAndGet(ref.sizeEstimate/2); + ref.sizeEstimate = 0; + ref.clear(); + } + } + @Override public int getProcessorBatchSize(List<? extends Expression> schema) { return getSizeEstimates(schema)[0]; @@ -821,9 +688,9 @@ boolean less = totalCopy < targetBytesPerRow; int rowCount = processorBatchSize; - for (int i = 0; i < 2; i++) { + for (int i = 0; i < 3; i++) { if (less) { - totalCopy <<= 2; + totalCopy <<= 1; } else { totalCopy >>= 2; } @@ -891,9 +758,13 @@ public void setInlineLobs(boolean inlineLobs) { this.inlineLobs = inlineLobs; } + + private int getMaxReserveKB() { + return maxReserveKB.get(); + } - public void setCompactionThreshold(int compactionThreshold) { - this.compactionThreshold = compactionThreshold; + public void setCache(Cache cache) { + this.cache = cache; } } Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/FileStorageManager.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/impl/FileStorageManager.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/FileStorageManager.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -100,60 +100,72 @@ this.name = name; } - public synchronized int readDirect(long fileOffset, byte[] b, int offSet, int length) throws TeiidComponentException { - try { - RandomAccessFile fileAccess = fileInfo.open(); - fileAccess.seek(fileOffset); - return fileAccess.read(b, offSet, length); - } catch (IOException e) { - throw new TeiidComponentException(e, QueryPlugin.Util.getString("FileStoreageManager.error_reading", fileInfo.file.getAbsoluteFile())); //$NON-NLS-1$ - } finally { - fileInfo.close(); - } + @Override + public synchronized long getLength() { + if (fileInfo == null) { + return 0; + } + return fileInfo.file.length(); } - - /** - * Concurrent writes are prevented by FileStore, but in general should not happen since processing is single threaded. - */ - public void writeDirect(long fileOffset, byte[] bytes, int offset, int length) throws TeiidComponentException { - long used = usedBufferSpace.addAndGet(length); - if (used > maxBufferSpace) { - usedBufferSpace.addAndGet(-length); - //TODO: trigger a compaction before this is thrown - throw new TeiidComponentException(QueryPlugin.Util.getString("FileStoreageManager.space_exhausted", maxBufferSpace)); //$NON-NLS-1$ - } + + @Override + protected synchronized int readWrite(long fileOffset, byte[] b, int offSet, + int length, boolean write) throws IOException { + if (!write) { + try { + RandomAccessFile fileAccess = fileInfo.open(); + fileAccess.seek(fileOffset); + return fileAccess.read(b, offSet, length); + } finally { + fileInfo.close(); + } + } if (fileInfo == null) { fileInfo = new FileInfo(createFile(name)); } + long bytesUsed = 0; try { RandomAccessFile fileAccess = fileInfo.open(); - fileAccess.setLength(fileOffset + length); + long newLength = fileOffset + length; + bytesUsed = newLength - fileAccess.length(); + if (bytesUsed > 0) { + long used = usedBufferSpace.addAndGet(bytesUsed); + if (used > maxBufferSpace) { + usedBufferSpace.addAndGet(-bytesUsed); + //TODO: trigger a compaction before this is thrown + throw new IOException(QueryPlugin.Util.getString("FileStoreageManager.space_exhausted", maxBufferSpace)); //$NON-NLS-1$ + } + fileAccess.setLength(bytesUsed); + bytesUsed = 0; + } fileAccess.seek(fileOffset); - fileAccess.write(bytes, offset, length); - } catch(IOException e) { - throw new TeiidComponentException(e, QueryPlugin.Util.getString("FileStoreageManager.error_reading", fileInfo.file.getAbsoluteFile())); //$NON-NLS-1$ + fileAccess.write(b, offSet, length); } finally { + if (bytesUsed > 0) { + usedBufferSpace.addAndGet(-bytesUsed); + } fileInfo.close(); - } - } + } + return length; + } + + @Override + public synchronized void setLength(long length) throws IOException { + try { + fileInfo.open().setLength(length); + } finally { + fileInfo.close(); + } + } - public void removeDirect() { - usedBufferSpace.addAndGet(-len); + @Override + public synchronized void removeDirect() { + usedBufferSpace.addAndGet(-getLength()); if (fileInfo != null){ fileInfo.delete(); } } - @Override - protected void truncateDirect(long length) throws TeiidComponentException { - try { - RandomAccessFile raf = fileInfo.open(); - raf.setLength(length); - } catch (IOException e) { - throw new TeiidComponentException(e, QueryPlugin.Util.getString("FileStoreageManager.error_reading", fileInfo.file.getAbsoluteFile())); //$NON-NLS-1$ - } - } - } // Initialization @@ -204,16 +216,12 @@ this.directory = directory; } - File createFile(String name) throws TeiidComponentException { - try { - File storageFile = File.createTempFile(FILE_PREFIX + name + "_", null, this.dirFile); //$NON-NLS-1$ - if (LogManager.isMessageToBeRecorded(org.teiid.logging.LogConstants.CTX_BUFFER_MGR, MessageLevel.DETAIL)) { - LogManager.logDetail(org.teiid.logging.LogConstants.CTX_BUFFER_MGR, "Created temporary storage area file " + storageFile.getAbsoluteFile()); //$NON-NLS-1$ - } - return storageFile; - } catch(IOException e) { - throw new TeiidComponentException(e, QueryPlugin.Util.getString("FileStoreageManager.error_creating", name)); //$NON-NLS-1$ + File createFile(String name) throws IOException { + File storageFile = File.createTempFile(FILE_PREFIX + name + "_", null, this.dirFile); //$NON-NLS-1$ + if (LogManager.isMessageToBeRecorded(org.teiid.logging.LogConstants.CTX_BUFFER_MGR, MessageLevel.DETAIL)) { + LogManager.logDetail(org.teiid.logging.LogConstants.CTX_BUFFER_MGR, "Created temporary storage area file " + storageFile.getAbsoluteFile()); //$NON-NLS-1$ } + return storageFile; } public FileStore createFileStore(String name) { Added: trunk/engine/src/main/java/org/teiid/common/buffer/impl/FileStoreCache.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/impl/FileStoreCache.java (rev 0) +++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/FileStoreCache.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -0,0 +1,417 @@ +/* + * JBoss, Home of Professional Open Source. + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301 USA. + */ + +package org.teiid.common.buffer.impl; + +import java.io.IOException; +import java.io.InputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +import org.teiid.common.buffer.Cache; +import org.teiid.common.buffer.CacheEntry; +import org.teiid.common.buffer.FileStore; +import org.teiid.common.buffer.Serializer; +import org.teiid.common.buffer.StorageManager; +import org.teiid.core.TeiidComponentException; +import org.teiid.core.TeiidRuntimeException; +import org.teiid.logging.LogConstants; +import org.teiid.logging.LogManager; +import org.teiid.logging.MessageLevel; +import org.teiid.query.QueryPlugin; + +/** + * Implements storage and caching against a {@link FileStore} abstraction. + * caching uses a block paradigm, where the first block has the format + * long gid | long oid | short tail bytes | short sector table length | [3 byte int sector ...] + * + * TODO: back non-fragmented block information onto the block address + */ +public class FileStoreCache implements Cache, StorageManager { + + private static final int IO_BUFFER_SIZE = 1 << 14; + private static final int DATA_START = 1 << 13; + private static final int MAX_BLOCKS = 2724; //implies that the max size is ~ 42.5 mb + + static class TryFreeParameter { + Long gid; + Long oid; + Integer block; + + public TryFreeParameter(Long gid, Long oid, Integer block) { + this.gid = gid; + this.oid = oid; + this.block = block; + } + } + + private class Segment { + int id; + ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); + List<TryFreeParameter> freed = Collections.synchronizedList(new ArrayList<TryFreeParameter>()); + FileStore store; + BitSetTree inuse = new BitSetTree(); + BitSetTree fragmentedFlags = new BitSetTree(); + + public Segment(int id) { + this.id = id; + } + + void tryFree(Long gid, Long oid, Integer block) { + if (lock.writeLock().tryLock()) { + try { + free(gid, oid, block); + } finally { + lock.writeLock().unlock(); + } + } else { + freed.add(new TryFreeParameter(gid, oid, block)); + } + } + + private void free(Long gid, Long oid, Integer block) { + if (block == null) { + Map<Long, Integer> map = physicalMapping.get(gid); + if (map == null) { + return; + } + block = map.remove(oid); + if (block == null) { + return; + } + } + byte[] buf = new byte[DATA_START]; + try { + inuse.set(block, false); + boolean fragmented = false; + if (fragmentedFlags.get(block)) { + fragmented = true; + store.read(block << 14, buf, 0, buf.length); + fragmentedFlags.set(block, false); + } else { + //TODO: if not fragmented come up with a better approach than a disk read + store.read(block << 14, buf, 0, 20); + } + ByteBuffer bb = ByteBuffer.wrap(buf); + bb.position(18); + short blocks = bb.getShort(); + if (!fragmented) { + for (int i = block + 1; i < blocks; i++) { + inuse.set(i, false); + } + } else { + for (short i = 0; i < blocks; i++) { + int toFree = getNextBlock(bb); + inuse.set(toFree, false); + } + } + } catch (IOException e) { + throw new TeiidRuntimeException(e, "Could not read intial block to process freeing " + oid); //$NON-NLS-1$ + } + } + + @SuppressWarnings("unchecked") + void add(final CacheEntry entry, final Serializer s) throws TeiidComponentException { + lock.writeLock().lock(); + try { + List<TryFreeParameter> toFree = Collections.emptyList(); + synchronized (freed) { + if (!freed.isEmpty()) { + toFree = new ArrayList<TryFreeParameter>(freed); + freed.clear(); + } + } + for (TryFreeParameter param : toFree) { + free(param.gid, param.oid, param.block); + } + ExtensibleBufferedOutputStream fsos = new ExtensibleBufferedOutputStream(new byte[IO_BUFFER_SIZE]) { + + byte[] firstBytes = null; + List<Integer> blocks = new ArrayList<Integer>(8); + boolean isFragmented; + int lastCount; + int start; + + @Override + protected void flushDirect() throws IOException { + lastCount = count; + if (firstBytes == null) { + start = nextBlock(-1); + firstBytes = buf; + //need a new buffer only if there could be bytes remaining + if (count == buf.length) { + buf = new byte[IO_BUFFER_SIZE]; + } + } else { + int last = -1; + if (!blocks.isEmpty()) { + last = blocks.get(blocks.size()-1); + } else { + last = start; + } + int next = nextBlock(last); + blocks.add(next); + if (next != last + 1) { + isFragmented = true; + } + if (blocks.size() > MAX_BLOCKS) { + //TODO handle this case + throw new TeiidRuntimeException("Exceeded max persistent object size" + entry.getId()); //$NON-NLS-1$ + } + store.write(next << 14, buf, 0, count); + } + } + + @Override + public void close() throws IOException { + super.close(); + ByteBuffer bb = ByteBuffer.wrap(firstBytes); + bb.putLong(s.getId()); + bb.putLong(entry.getId()); + bb.putShort((short)(lastCount - (blocks.isEmpty()?DATA_START:0))); + bb.putShort((short)blocks.size()); + for (Integer i : blocks) { + bb.put((byte)(i.intValue()>>16)); + bb.putShort((short)i.intValue()); + } + store.write(start << 14, firstBytes, 0, firstBytes.length); + if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.TRACE)) { + LogManager.logTrace(LogConstants.CTX_BUFFER_MGR, id, entry.getId(), "batch written starting at:", start); //$NON-NLS-1$ + } + Map<Long, Integer> map = physicalMapping.get(s.getId()); + if (map == null) { + return; + } + map.put(entry.getId(), start); + inuse.set(start, true); + for (Integer i : blocks) { + inuse.set(i, true); + } + if (isFragmented) { + fragmentedFlags.set(start, true); + } + } + + }; + fsos.count = DATA_START; + ObjectOutputStream oos = new ObjectOutputStream(fsos); + oos.writeInt(entry.getSizeEstimate()); + s.serialize(entry.getObject(), oos); + oos.close(); + } catch (IOException e) { + throw new TeiidComponentException(e); + } finally { + lock.writeLock().unlock(); + } + + if (entry.getId().longValue() == 451) { + get(entry.getId(), s); + } + } + + int nextBlock(int fromIndex) { + int next = inuse.nextClearBit(fromIndex + 1); + if (next == -1) { + throw new TeiidRuntimeException("no freespace available on segment" + id); //$NON-NLS-1$ + } + return next; + } + + @SuppressWarnings("unchecked") + CacheEntry get(Long oid, Serializer serializer) throws TeiidComponentException { + lock.readLock().lock(); + try { + Map<Long, Integer> map = physicalMapping.get(serializer.getId()); + if (map == null) { + return null; + } + final Integer startBlock = map.get(oid); + if (startBlock == null) { + return null; + } + ObjectInputStream ois = new ObjectInputStream(new InputStream() { + + int count; + int pos; + byte[] buf = new byte[IO_BUFFER_SIZE]; + ByteBuffer firstBytes; + int currentBlock; + short tailBytes; + short totalBlocks; + short blockNum; + + @Override + public int read() throws IOException { + if (pos == count) { + if (firstBytes == null) { + store.readFully(startBlock << 14, buf, 0, buf.length); + firstBytes = ByteBuffer.wrap(buf); + firstBytes.position(16); + tailBytes = firstBytes.getShort(); + totalBlocks = firstBytes.getShort(); + count = buf.length; + pos = DATA_START; + } else { + buf = new byte[IO_BUFFER_SIZE]; + //TODO: defrag on read + if (count == buf.length) { + currentBlock = getNextBlock(firstBytes); + blockNum++; + pos = 0; + } + int length = blockNum == totalBlocks?tailBytes:buf.length; + store.readFully(currentBlock << 14, buf, 0, length); + count = length; + } + } + return buf[pos++] & 0xff; + } + }); + CacheEntry ce = new CacheEntry(oid); + ce.setSizeEstimate(ois.readInt()); + ce.setObject(serializer.deserialize(ois)); + ce.setPersistent(true); + return ce; + } catch(IOException e) { + throw new TeiidComponentException(e, QueryPlugin.Util.getString("FileStoreageManager.error_reading", id)); //$NON-NLS-1$ + } catch (ClassNotFoundException e) { + throw new TeiidComponentException(e, QueryPlugin.Util.getString("FileStoreageManager.error_reading", id)); //$NON-NLS-1$ + } finally { + lock.readLock().unlock(); + } + } + } + + static int getNextBlock(ByteBuffer bb) { + int block = (bb.get() & 0xff)<< 16; + block += (bb.getShort() & 0xffff); + return block; + } + + private StorageManager storageManager; + private Segment[] segments; + private ConcurrentHashMap<Long, Map<Long, Integer>> physicalMapping = new ConcurrentHashMap<Long, Map<Long,Integer>>(); + private int segmentCount = 32; + + @Override + public void add(CacheEntry entry, Serializer<?> s) { + Segment seg = getSegment(entry.getId()); + try { + seg.add(entry, s); + } catch (Throwable e) { + LogManager.logError(LogConstants.CTX_BUFFER_MGR, e, "Error persisting batch, attempts to read batch "+ entry.getId() +" later will result in an exception"); //$NON-NLS-1$ //$NON-NLS-2$ + } + } + + @Override + public CacheEntry get(Long id, Serializer<?> serializer) throws TeiidComponentException { + Segment seg = getSegment(id); + return seg.get(id, serializer); + } + + private Segment getSegment(Long id) { + return segments[id.hashCode() & (segments.length - 1)]; + } + + @Override + public FileStore createFileStore(String name) { + return storageManager.createFileStore(name); + } + + @Override + public void initialize() throws TeiidComponentException { + storageManager.initialize(); + int numSegments = 31 - Integer.numberOfLeadingZeros(this.segmentCount); + segments = new Segment[1 << numSegments]; + for (int i = 0; i < segments.length; i++) { + segments[i] = new Segment(i); + segments[i].store = storageManager.createFileStore("segment_" + i); //$NON-NLS-1$ + } + } + + @Override + public void addToCacheGroup(Long gid, Long oid) { + Map<Long, Integer> map = physicalMapping.get(gid); + if (map == null) { + return; + } + map.put(oid, null); + } + + @Override + public void createCacheGroup(Long gid) { + physicalMapping.put(gid, Collections.synchronizedMap(new HashMap<Long, Integer>())); + } + + @Override + public void remove(Long gid, Long id) { + Map<Long, Integer> map = physicalMapping.get(gid); + if (map == null) { + return; + } + Integer block = map.remove(id); + if (block != null) { + Segment s = getSegment(id); + s.tryFree(gid, id, block); + } + } + + @Override + public Collection<Long> removeCacheGroup(Long gid) { + Map<Long, Integer> values = physicalMapping.remove(gid); + if (values == null) { + return Collections.emptySet(); + } + synchronized (values) { + for (Map.Entry<Long, Integer> entry : values.entrySet()) { + if (entry.getValue() != null) { + Segment s = getSegment(entry.getKey()); + s.tryFree(gid, entry.getKey(), entry.getValue()); + } + } + return new HashSet<Long>(values.keySet()); + } + } + + public void setSegmentCount(int segmentCount) { + this.segmentCount = segmentCount; + } + + public void setStorageManager(StorageManager storageManager) { + this.storageManager = storageManager; + } + + public StorageManager getStorageManager() { + return storageManager; + } + +} Property changes on: trunk/engine/src/main/java/org/teiid/common/buffer/impl/FileStoreCache.java ___________________________________________________________________ Added: svn:mime-type + text/plain Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/MemoryStorageManager.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/impl/MemoryStorageManager.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/MemoryStorageManager.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -23,70 +23,85 @@ package org.teiid.common.buffer.impl; import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; +import org.teiid.common.buffer.Cache; +import org.teiid.common.buffer.CacheEntry; import org.teiid.common.buffer.FileStore; -import org.teiid.common.buffer.StorageManager; +import org.teiid.common.buffer.Serializer; import org.teiid.core.TeiidComponentException; -public class MemoryStorageManager implements StorageManager { +public class MemoryStorageManager implements Cache { + + public static final int MAX_FILE_SIZE = 1 << 17; - private AtomicInteger created = new AtomicInteger(); - private AtomicInteger removed = new AtomicInteger(); - - public void initialize() { - } + private final class MemoryFileStore extends FileStore { + private ByteBuffer buffer = ByteBuffer.allocate(MAX_FILE_SIZE); + + public MemoryFileStore() { + buffer.limit(0); + } - @Override - public FileStore createFileStore(String name) { - created.incrementAndGet(); - return new FileStore() { - private ByteBuffer buffer = ByteBuffer.allocate(1 << 16); - - @Override - public void writeDirect(long start, byte[] bytes, int offset, int length) throws TeiidComponentException { - buffer.position((int)start); - if (buffer.position() + length > buffer.capacity()) { - ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2 + length); - buffer.position(0); - newBuffer.put(buffer); - buffer = newBuffer; - buffer.position((int)start); - } - buffer.put(bytes, offset, length); - } - - @Override - public void removeDirect() { - removed.incrementAndGet(); - buffer = ByteBuffer.allocate(0); - } - - @Override - public synchronized int readDirect(long fileOffset, byte[] b, int offset, int length) - throws TeiidComponentException { + @Override + public synchronized void removeDirect() { + removed.incrementAndGet(); + buffer = ByteBuffer.allocate(0); + } + + @Override + protected synchronized int readWrite(long fileOffset, byte[] b, int offSet, + int length, boolean write) { + if (!write) { if (fileOffset >= getLength()) { return -1; } int position = (int)fileOffset; buffer.position(position); length = Math.min(length, (int)getLength() - position); - buffer.get(b, offset, length); - return length; + buffer.get(b, offSet, length); + return length; } - - @Override - protected void truncateDirect(long length) { - ByteBuffer newBuffer = ByteBuffer.allocate((int)length); - buffer.position(0); - buffer.limit(newBuffer.capacity()); - newBuffer.put(buffer); - buffer = newBuffer; + int requiredLength = (int)(fileOffset + length); + if (requiredLength > buffer.limit()) { + buffer.limit(requiredLength); } - }; + buffer.position((int)fileOffset); + buffer.put(b, offSet, length); + return length; + } + + @Override + public synchronized void setLength(long length) { + buffer.limit((int)length); + } + + @Override + public synchronized long getLength() { + return buffer.limit(); + } + } + + private Map<Long, Map<Long, CacheEntry>> groups = new ConcurrentHashMap<Long, Map<Long, CacheEntry>>(); + private AtomicInteger created = new AtomicInteger(); + private AtomicInteger removed = new AtomicInteger(); + public void initialize() { + } + + @Override + public FileStore createFileStore(String name) { + created.incrementAndGet(); + return new MemoryFileStore(); + } + public int getCreated() { return created.get(); } @@ -94,4 +109,55 @@ public int getRemoved() { return removed.get(); } + + @Override + public void add(CacheEntry entry, Serializer<?> s) { + Map<Long, CacheEntry> group = groups.get(s.getId()); + if (group != null) { + group.put(entry.getId(), entry); + } + } + + @Override + public void addToCacheGroup(Long gid, Long oid) { + Map<Long, CacheEntry> group = groups.get(gid); + if (group != null) { + group.put(oid, null); + } + } + + @Override + public void createCacheGroup(Long gid) { + groups.put(gid, Collections.synchronizedMap(new HashMap<Long, CacheEntry>())); + } + + @Override + public CacheEntry get(Long id, Serializer<?> serializer) + throws TeiidComponentException { + Map<Long, CacheEntry> group = groups.get(id); + if (group != null) { + return group.get(id); + } + return null; + } + + @Override + public void remove(Long gid, Long id) { + Map<Long, CacheEntry> group = groups.get(gid); + if (group != null) { + group.remove(id); + } + } + + @Override + public Collection<Long> removeCacheGroup(Long gid) { + Map<Long, CacheEntry> group = groups.remove(gid); + if (group == null) { + return Collections.emptySet(); + } + synchronized (group) { + return new ArrayList<Long>(group.keySet()); + } + } + } \ No newline at end of file Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/SizeUtility.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/impl/SizeUtility.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/SizeUtility.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -26,9 +26,9 @@ import java.math.BigDecimal; import java.math.BigInteger; import java.util.HashMap; +import java.util.List; import java.util.Map; -import org.teiid.common.buffer.TupleBatch; import org.teiid.core.types.DataTypeManager; @@ -65,27 +65,29 @@ private long bigIntegerEstimate; private long bigDecimalEstimate; + private String[] types; - public SizeUtility() { + public SizeUtility(String[] types) { boolean isValueCacheEnabled = DataTypeManager.isValueCacheEnabled(); bigIntegerEstimate = getSize(isValueCacheEnabled, DataTypeManager.DefaultDataClasses.BIG_INTEGER); bigDecimalEstimate = getSize(isValueCacheEnabled, DataTypeManager.DefaultDataClasses.BIG_DECIMAL); + this.types = types; } - public long getBatchSize(TupleBatch data) { + public long getBatchSize(List<? extends List<?>> data) { return getBatchSize(DataTypeManager.isValueCacheEnabled(), data); } - private long getBatchSize(boolean accountForValueCache, TupleBatch data) { - int colLength = data.getDataTypes().length; - int rowLength = data.getRowCount(); + private long getBatchSize(boolean accountForValueCache, List<? extends List<?>> data) { + int colLength = types.length; + int rowLength = data.size(); // Array overhead for row array long size = 16 + alignMemory(rowLength * REFERENCE_SIZE); // array overhead for all the columns ( 8 object overhead + 4 ref + 4 int) size += (rowLength * (48 + alignMemory(colLength * REFERENCE_SIZE))); for (int col = 0; col < colLength; col++) { - Class<?> type = DataTypeManager.getDataTypeClass(data.getDataTypes()[col]); + Class<?> type = DataTypeManager.getDataTypeClass(types[col]); if (type == DataTypeManager.DefaultDataClasses.STRING || type == DataTypeManager.DefaultDataClasses.OBJECT @@ -94,7 +96,7 @@ int estRow = 0; for (int row = 0; row < rowLength; row++) { boolean updateEst = row == estRow; - size += getSize(data.getTuples().get(row).get(col), updateEst, accountForValueCache); + size += getSize(data.get(row).get(col), updateEst, accountForValueCache); if (updateEst) { estRow = estRow * 2 + 1; } Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/SplittableStorageManager.java =================================================================== --- trunk/engine/src/main/java/org/teiid/common/buffer/impl/SplittableStorageManager.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/SplittableStorageManager.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -22,9 +22,9 @@ package org.teiid.common.buffer.impl; -import java.util.Iterator; -import java.util.Map; -import java.util.concurrent.ConcurrentSkipListMap; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; import org.teiid.common.buffer.FileStore; import org.teiid.common.buffer.StorageManager; @@ -32,8 +32,8 @@ public class SplittableStorageManager implements StorageManager { - public static final long DEFAULT_MAX_FILESIZE = 2L * 1024L; - private long maxFileSize = DEFAULT_MAX_FILESIZE * 1024L * 1024L; // 2GB + public static final long DEFAULT_MAX_FILESIZE = 2 * 1024l; + private long maxFileSize = DEFAULT_MAX_FILESIZE * 1024l * 1024l; // 2GB private StorageManager storageManager; public SplittableStorageManager(StorageManager storageManager) { @@ -52,73 +52,93 @@ public class SplittableFileStore extends FileStore { private String name; - private ConcurrentSkipListMap<Long, FileStore> storageFiles = new ConcurrentSkipListMap<Long, FileStore>(); + private List<FileStore> storageFiles = new ArrayList<FileStore>(); + + private volatile long len; public SplittableFileStore(String name) { this.name = name; } @Override - public int readDirect(long fileOffset, byte[] b, int offSet, int length) throws TeiidComponentException { - Map.Entry<Long, FileStore> entry = storageFiles.floorEntry(fileOffset); - FileStore fileInfo = entry.getValue(); - return fileInfo.read(fileOffset - entry.getKey(), b, offSet, length); + public long getLength() { + return len; } + + @Override + protected int readWrite(long fileOffset, byte[] b, int offSet, + int length, boolean write) throws IOException { + FileStore store = null; + if (!write) { + synchronized (this) { + if (fileOffset + length > len) { + throw new IOException("Invalid file position " + fileOffset + " length " + length); //$NON-NLS-1$ //$NON-NLS-2$ + } + store = storageFiles.get((int)(fileOffset/maxFileSize)); + } + return store.read(fileOffset%maxFileSize, b, offSet, length); + } + synchronized (this) { + ensureLength(fileOffset + length); + store = storageFiles.get((int)(fileOffset/maxFileSize)); + } + long fileBegin = (int)(fileOffset%maxFileSize); + length = Math.min(length, (int)Math.min(Integer.MAX_VALUE, maxFileSize - fileBegin)); + store.write(fileBegin, b, offSet, length); + return length; + } + private void ensureLength(long length) throws IOException { + if (length <= len) { + return; + } + int numFiles = (int)(length/maxFileSize); + long lastFileSize = length%maxFileSize; + if (lastFileSize > 0) { + numFiles++; + } + for (int i = storageFiles.size(); i < numFiles; i++) { + FileStore newFileInfo = storageManager.createFileStore(name + "_" + storageFiles.size()); //$NON-NLS-1$ + storageFiles.add(newFileInfo); + if (lastFileSize == 0 || i != numFiles - 1) { + newFileInfo.setLength(maxFileSize); + } + } + if (lastFileSize > 0) { + storageFiles.get(storageFiles.size() - 1).setLength(lastFileSize); + } + len = length; + } + @Override - public void writeDirect(long start, byte[] bytes, int offset, int length) throws TeiidComponentException { - Map.Entry<Long, FileStore> entry = this.storageFiles.floorEntry(start); - boolean createNew = false; - FileStore fileInfo = null; - long fileOffset = 0; - if (entry == null) { - createNew = true; + public synchronized void setLength(long length) throws IOException { + if (length > len) { + ensureLength(length); } else { - fileInfo = entry.getValue(); - fileOffset = entry.getKey(); - if (start > entry.getValue().getLength() + fileOffset) { - throw new AssertionError("invalid write start location"); //$NON-NLS-1$ + int numFiles = (int)(length/maxFileSize); + long lastFileSize = length%maxFileSize; + if (lastFileSize > 0) { + numFiles++; } - createNew = start + length > getMaxFileSize(); + int toRemove = storageFiles.size() - numFiles; + for (int i = 0; i < toRemove; i++) { + FileStore store = storageFiles.remove(storageFiles.size() -1); + store.remove(); + } + if (lastFileSize > 0) { + storageFiles.get(storageFiles.size() - 1).setLength(lastFileSize); + } } - if (createNew) { - FileStore newFileInfo = storageManager.createFileStore(name + "_" + storageFiles.size()); //$NON-NLS-1$ - if (fileInfo != null) { - fileOffset += fileInfo.getLength(); - } - storageFiles.put(fileOffset, newFileInfo); - fileInfo = newFileInfo; - } - fileInfo.write(start - fileOffset, bytes, offset, length); - } - - public void removeDirect() { - for (FileStore info : storageFiles.values()) { + len = length; + } + + public synchronized void removeDirect() { + for (FileStore info : storageFiles) { info.remove(); } storageFiles.clear(); } - @Override - protected void truncateDirect(long length) - throws TeiidComponentException { - Map.Entry<Long, FileStore> start = storageFiles.floorEntry(length); - if (start == null) { - return; - } - if (start.getKey().longValue() == length) { - start.getValue().remove(); - storageFiles.remove(start.getKey()); - } else { - start.getValue().truncate(length - start.getKey()); - } - for (Iterator<FileStore> iter = storageFiles.tailMap(length, false).values().iterator(); iter.hasNext();) { - iter.next().remove(); - iter.remove(); - } - - } - } public long getMaxFileSize() { @@ -126,10 +146,10 @@ } public void setMaxFileSize(long maxFileSize) { - this.maxFileSize = maxFileSize * 1024L * 1024L; + this.maxFileSize = maxFileSize * 1024l * 1024l; } - void setMaxFileSizeDirect(long maxFileSize) { + public void setMaxFileSizeDirect(long maxFileSize) { this.maxFileSize = maxFileSize; } Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorWorkItem.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorWorkItem.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorWorkItem.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -31,7 +31,6 @@ import org.teiid.adminapi.impl.VDBMetaData; import org.teiid.common.buffer.BlockedException; -import org.teiid.common.buffer.TupleBuffer; import org.teiid.core.TeiidProcessingException; import org.teiid.core.util.Assertion; import org.teiid.dqp.message.AtomicRequestID; @@ -332,7 +331,7 @@ LogManager.logWarning(LogConstants.CTX_CONNECTOR, QueryPlugin.Util.getString("ConnectorWorker.zero_size_non_last_batch", requestMsg.getConnectorName())); //$NON-NLS-1$ } - AtomicResultsMessage response = createResultsMessage(rows.toArray(new List[currentRowCount]), requestMsg.getCommand().getProjectedSymbols()); + AtomicResultsMessage response = createResultsMessage(rows.toArray(new List[currentRowCount])); // if we need to keep the execution alive, then we can not support implicit close. response.setSupportsImplicitClose(!this.securityContext.keepExecutionAlive()); @@ -346,9 +345,8 @@ return response; } - public static AtomicResultsMessage createResultsMessage(List[] batch, List columnSymbols) { - String[] dataTypes = TupleBuffer.getTypeNames(columnSymbols); - return new AtomicResultsMessage(batch, dataTypes); + public static AtomicResultsMessage createResultsMessage(List[] batch) { + return new AtomicResultsMessage(batch); } boolean isCancelled() { Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -398,7 +398,7 @@ fullyCloseSource(); } if(workItem.requestMsg.supportsPartialResults()) { - AtomicResultsMessage emptyResults = new AtomicResultsMessage(new List[0], null); + AtomicResultsMessage emptyResults = new AtomicResultsMessage(new List[0]); emptyResults.setWarnings(Arrays.asList((Exception)exception)); emptyResults.setFinalRow(this.rowsProcessed); return emptyResults; Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -25,12 +25,13 @@ import java.sql.ResultSet; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.concurrent.Callable; -import java.util.concurrent.ConcurrentHashMap; import org.teiid.client.RequestMessage; import org.teiid.client.ResultsMessage; @@ -144,7 +145,7 @@ * maintained during processing */ private Throwable processingException; - private Map<AtomicRequestID, DataTierTupleSource> connectorInfo = new ConcurrentHashMap<AtomicRequestID, DataTierTupleSource>(4); + private Map<AtomicRequestID, DataTierTupleSource> connectorInfo = Collections.synchronizedMap(new HashMap<AtomicRequestID, DataTierTupleSource>(4)); // This exception contains details of all the atomic requests that failed when query is run in partial results mode. private List<TeiidException> warnings = new LinkedList<TeiidException>(); private volatile boolean doneProducingBatches; @@ -157,7 +158,7 @@ private int begin; private int end; private TupleBatch savedBatch; - private Map<Integer, LobWorkItem> lobStreams = new ConcurrentHashMap<Integer, LobWorkItem>(4); + private Map<Integer, LobWorkItem> lobStreams = Collections.synchronizedMap(new HashMap<Integer, LobWorkItem>(4)); /**The time when command begins processing on the server.*/ private long processingTimestamp = System.currentTimeMillis(); @@ -362,7 +363,7 @@ * break the read of a lob from a transactional source under a transaction * if the source does not support holding the clob open after commit */ - for (DataTierTupleSource connectorRequest : this.connectorInfo.values()) { + for (DataTierTupleSource connectorRequest : getConnectorRequests()) { if (connectorRequest.isTransactional()) { connectorRequest.fullyCloseSource(); } @@ -401,15 +402,21 @@ } } - for (DataTierTupleSource connectorRequest : this.connectorInfo.values()) { + for (DataTierTupleSource connectorRequest : getConnectorRequests()) { connectorRequest.fullyCloseSource(); } } this.resultsBuffer = null; - for (LobWorkItem lobWorkItem : this.lobStreams.values()) { - lobWorkItem.close(); + if (!this.lobStreams.isEmpty()) { + List<LobWorkItem> lobs = null; + synchronized (lobStreams) { + lobs = new ArrayList<LobWorkItem>(this.lobStreams.values()); + } + for (LobWorkItem lobWorkItem : lobs) { + lobWorkItem.close(); + } } } @@ -592,7 +599,7 @@ } int finalRowCount = this.resultsBuffer.isFinal()?this.resultsBuffer.getRowCount():(batch.getTerminationFlag()?batch.getEndRow():-1); - response = createResultsMessage(batch.getAllTuples(), this.originalCommand.getProjectedSymbols()); + response = createResultsMessage(batch.getTuples(), this.originalCommand.getProjectedSymbols()); response.setFirstRow(batch.getBeginRow()); response.setLastRow(batch.getEndRow()); response.setUpdateResult(this.returnsUpdateCount); @@ -631,7 +638,7 @@ return result; } - public ResultsMessage createResultsMessage(List[] batch, List columnSymbols) { + public ResultsMessage createResultsMessage(List<? extends List<?>> batch, List columnSymbols) { String[] columnNames = new String[columnSymbols.size()]; String[] dataTypes = new String[columnSymbols.size()]; @@ -640,7 +647,7 @@ columnNames[i] = SingleElementSymbol.getShortName(symbol.getOutputName()); dataTypes[i] = DataTypeManager.getDataTypeName(symbol.getType()); } - ResultsMessage result = new ResultsMessage(requestMsg, batch, columnNames, dataTypes); + ResultsMessage result = new ResultsMessage(batch, columnNames, dataTypes); setAnalysisRecords(result); return result; } @@ -672,7 +679,7 @@ } } LogManager.logDetail(LogConstants.CTX_DQP, processingException, "Sending error to client", requestID); //$NON-NLS-1$ - ResultsMessage response = new ResultsMessage(requestMsg); + ResultsMessage response = new ResultsMessage(); Throwable exception = this.processingException; if (isCanceled) { exception = addCancelCode(exception); @@ -737,7 +744,7 @@ // Cancel Connector atomic requests try { - for (DataTierTupleSource connectorRequest : this.connectorInfo.values()) { + for (DataTierTupleSource connectorRequest : getConnectorRequests()) { connectorRequest.cancelRequest(); } } finally { @@ -837,7 +844,9 @@ Collection<DataTierTupleSource> getConnectorRequests() { - return new LinkedList<DataTierTupleSource>(this.connectorInfo.values()); + synchronized (this.connectorInfo) { + return new ArrayList<DataTierTupleSource>(this.connectorInfo.values()); + } } DataTierTupleSource getConnectorRequest(AtomicRequestID id) { @@ -874,7 +883,7 @@ this.doneProducingBatches = true; //TODO: we could perform more tracking to know what source lobs are in use if (this.resultsBuffer.getLobCount() == 0) { - for (DataTierTupleSource connectorRequest : this.connectorInfo.values()) { + for (DataTierTupleSource connectorRequest : getConnectorRequests()) { connectorRequest.fullyCloseSource(); } } Added: trunk/engine/src/main/java/org/teiid/dqp/internal/process/SerializableTupleBatch.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/internal/process/SerializableTupleBatch.java (rev 0) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/SerializableTupleBatch.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -0,0 +1,65 @@ +/* + * JBoss, Home of Professional Open Source. + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301 USA. + */ + +package org.teiid.dqp.internal.process; + +import java.io.Externalizable; +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.util.List; + +import org.teiid.client.BatchSerializer; +import org.teiid.common.buffer.TupleBatch; +import org.teiid.core.util.ExternalizeUtil; + +public class SerializableTupleBatch extends TupleBatch implements Externalizable { + + private String[] types; + + public SerializableTupleBatch() { + //for Externalizable + } + + public SerializableTupleBatch(TupleBatch batch, String[] types) { + super(batch.getBeginRow(), batch.getTuples()); + this.types = types; + } + + @Override + public void readExternal(ObjectInput in) throws IOException, + ClassNotFoundException { + String[] types = ExternalizeUtil.readStringArray(in); + this.setRowOffset(in.readInt()); + this.setTerminationFlag(in.readBoolean()); + this.tuples = (List)BatchSerializer.readBatch(in, types); + } + + @Override + public void writeExternal(ObjectOutput out) throws IOException { + ExternalizeUtil.writeArray(out, types); + out.writeInt(this.getBeginRow()); + out.writeBoolean(this.getTerminationFlag()); + BatchSerializer.writeBatch(out, types, this.getTuples()); + } + +} Property changes on: trunk/engine/src/main/java/org/teiid/dqp/internal/process/SerializableTupleBatch.java ___________________________________________________________________ Added: svn:mime-type + text/plain Modified: trunk/engine/src/main/java/org/teiid/dqp/message/AtomicResultsMessage.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/message/AtomicResultsMessage.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/dqp/message/AtomicResultsMessage.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -22,20 +22,12 @@ package org.teiid.dqp.message; -import java.io.Externalizable; -import java.io.IOException; -import java.io.ObjectInput; -import java.io.ObjectOutput; import java.util.List; -import org.teiid.client.BatchSerializer; -import org.teiid.core.util.ExternalizeUtil; +public class AtomicResultsMessage { -public class AtomicResultsMessage implements Externalizable { - private List[] results; - private String[] dataTypes; // Final row index in complete result set, if known private int finalRow = -1; @@ -53,8 +45,7 @@ public AtomicResultsMessage() { } - public AtomicResultsMessage(List[] results, String[] dataTypes) { - this.dataTypes = dataTypes; + public AtomicResultsMessage(List[] results) { this.results = results; } @@ -86,26 +77,6 @@ return results; } - public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { - dataTypes = ExternalizeUtil.readStringArray(in); - results = BatchSerializer.readBatch(in, dataTypes); - finalRow = in.readInt(); - supportsImplicitClose = in.readBoolean(); - warnings = (List<Exception>)in.readObject(); - isTransactional = in.readBoolean(); - supportsCloseWithLobs = in.readBoolean(); - } - - public void writeExternal(ObjectOutput out) throws IOException { - ExternalizeUtil.writeArray(out, dataTypes); - BatchSerializer.writeBatch(out, dataTypes, results); - out.writeInt(finalRow); - out.writeBoolean(supportsImplicitClose); - out.writeObject(warnings); - out.writeBoolean(isTransactional); - out.writeBoolean(supportsCloseWithLobs); - } - public boolean isTransactional() { return isTransactional; } Modified: trunk/engine/src/main/java/org/teiid/query/processor/relational/EnhancedSortMergeJoinStrategy.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/processor/relational/EnhancedSortMergeJoinStrategy.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/query/processor/relational/EnhancedSortMergeJoinStrategy.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -141,6 +141,7 @@ List<?> lastTuple = null; boolean sortedDistinct = sorted && !state.isDistinct(); int sizeHint = index.getExpectedHeight(state.getTupleBuffer().getRowCount()); + index.setBatchInsert(sorted); outer: while (its.hasNext()) { //detect if sorted and distinct List<?> originalTuple = its.nextTuple(); @@ -162,6 +163,8 @@ } if (!sorted) { index.compact(); + } else { + index.setBatchInsert(false); } its.closeSource(); this.reverseIndexes = new int[elements.size()]; Modified: trunk/engine/src/main/java/org/teiid/query/tempdata/TempTable.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/tempdata/TempTable.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/main/java/org/teiid/query/tempdata/TempTable.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -90,6 +90,19 @@ this.addRowId = addRowId; this.indexes = indexes; } + + @Override + int process() throws ExpressionEvaluationException, + TeiidComponentException, TeiidProcessingException { + tree.setBatchInsert(addRowId); + return super.process(); + } + + @Override + protected void afterCompletion() throws TeiidComponentException { + tree.setBatchInsert(false); + } + @Override protected void tuplePassed(List tuple) throws BlockedException, @@ -237,6 +250,7 @@ success = true; } finally { try { + afterCompletion(); if (!success && undoLog != null) { undoLog.setFinal(true); TupleSource undoTs = undoLog.createIndexedTupleSource(); @@ -260,6 +274,14 @@ return updateCount; } + /** + * + * @throws TeiidComponentException + */ + protected void afterCompletion() throws TeiidComponentException { + + } + @SuppressWarnings("unused") void success() throws TeiidComponentException, ExpressionEvaluationException, TeiidProcessingException {} Modified: trunk/engine/src/test/java/org/teiid/common/buffer/BufferManagerFactory.java =================================================================== --- trunk/engine/src/test/java/org/teiid/common/buffer/BufferManagerFactory.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/test/java/org/teiid/common/buffer/BufferManagerFactory.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -22,9 +22,10 @@ package org.teiid.common.buffer; -import org.teiid.common.buffer.BufferManager; import org.teiid.common.buffer.impl.BufferManagerImpl; +import org.teiid.common.buffer.impl.FileStoreCache; import org.teiid.common.buffer.impl.MemoryStorageManager; +import org.teiid.common.buffer.impl.SplittableStorageManager; import org.teiid.core.TeiidComponentException; @@ -81,12 +82,18 @@ public static BufferManagerImpl initBufferManager(BufferManagerImpl bufferManager) { try { bufferManager.initialize(); - } catch (TeiidComponentException e) { + MemoryStorageManager storageManager = new MemoryStorageManager(); + SplittableStorageManager ssm = new SplittableStorageManager(storageManager); + ssm.setMaxFileSizeDirect(MemoryStorageManager.MAX_FILE_SIZE); + FileStoreCache fsc = new FileStoreCache(); + fsc.setSegmentCount(1); + fsc.setStorageManager(ssm); + fsc.initialize(); + bufferManager.setCache(fsc); + return bufferManager; + } catch (TeiidComponentException e) { throw new RuntimeException(e); } - - bufferManager.setStorageManager(new MemoryStorageManager()); - return bufferManager; } } Modified: trunk/engine/src/test/java/org/teiid/common/buffer/TestLobManager.java =================================================================== --- trunk/engine/src/test/java/org/teiid/common/buffer/TestLobManager.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/test/java/org/teiid/common/buffer/TestLobManager.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -28,8 +28,10 @@ import java.io.StringReader; import java.nio.charset.Charset; import java.util.Arrays; +import java.util.List; import org.junit.Test; +import org.teiid.common.buffer.LobManager.ReferenceMode; import org.teiid.common.buffer.impl.BufferManagerImpl; import org.teiid.core.types.BlobImpl; import org.teiid.core.types.BlobType; @@ -65,10 +67,11 @@ })); - LobManager lobManager = new LobManager(); + LobManager lobManager = new LobManager(new int[] {0, 1}, fs); lobManager.setMaxMemoryBytes(4); - lobManager.updateReferences(new int[] {0,1}, Arrays.asList(clob, blob)); - lobManager.persist(fs); + List<Streamable<? extends Object>> tuple = Arrays.asList(clob, blob); + lobManager.updateReferences(tuple, ReferenceMode.CREATE); + lobManager.persist(); Streamable<?>lob = lobManager.getLobReference(clob.getReferenceStreamId()); assertTrue(lob.getClass().isAssignableFrom(ClobType.class)); @@ -81,6 +84,10 @@ BlobType blobRead = (BlobType)lob; assertTrue(Arrays.equals(ObjectConverterUtil.convertToByteArray(blob.getBinaryStream()), ObjectConverterUtil.convertToByteArray(blobRead.getBinaryStream()))); + lobManager.updateReferences(tuple, ReferenceMode.REMOVE); + + assertEquals(0, lobManager.getLobCount()); + } } Modified: trunk/engine/src/test/java/org/teiid/common/buffer/TestSTree.java =================================================================== --- trunk/engine/src/test/java/org/teiid/common/buffer/TestSTree.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/test/java/org/teiid/common/buffer/TestSTree.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -109,7 +109,6 @@ BufferManagerImpl bm = BufferManagerFactory.createBufferManager(); bm.setProcessorBatchSize(32); bm.setMaxReserveKB(0);//force all to disk - bm.setCompactionThreshold(0); bm.initialize(); ElementSymbol e1 = new ElementSymbol("x"); Modified: trunk/engine/src/test/java/org/teiid/common/buffer/TestTupleBuffer.java =================================================================== --- trunk/engine/src/test/java/org/teiid/common/buffer/TestTupleBuffer.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/test/java/org/teiid/common/buffer/TestTupleBuffer.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -30,60 +30,18 @@ import javax.sql.rowset.serial.SerialClob; import org.junit.Test; -import org.mockito.Mockito; -import org.teiid.core.TeiidComponentException; +import org.teiid.common.buffer.BufferManager.TupleSourceType; import org.teiid.core.types.ClobType; import org.teiid.core.types.DataTypeManager; import org.teiid.query.sql.symbol.ElementSymbol; - public class TestTupleBuffer { - public static final class FakeBatchManager implements BatchManager { - @Override - public void remove() { - - } - - @Override - public ManagedBatch createManagedBatch(final TupleBatch batch, boolean softCache) - throws TeiidComponentException { - return new ManagedBatch() { - - @Override - public void remove() { - - } - - @Override - public TupleBatch getBatch(boolean cache, String[] types) - throws TeiidComponentException { - return batch; - } - - @Override - public void setPrefersMemory(boolean prefers) { - - } - - @Override - public CleanupHook getCleanupHook() { - return null; - } - }; - } - - @Override - public FileStore createStorage(String prefix) { - return Mockito.mock(FileStore.class); - } - } - @Test public void testForwardOnly() throws Exception { ElementSymbol x = new ElementSymbol("x"); //$NON-NLS-1$ x.setType(DataTypeManager.DefaultDataClasses.INTEGER); List<ElementSymbol> schema = Arrays.asList(x); - TupleBuffer tb = new TupleBuffer(new FakeBatchManager(), "x", schema, null, 32); //$NON-NLS-1$ + TupleBuffer tb = BufferManagerFactory.getStandaloneBufferManager().createTupleBuffer(schema, "x", TupleSourceType.PROCESSOR); //$NON-NLS-1$ tb.setForwardOnly(true); tb.addTuple(Arrays.asList(1)); TupleBatch batch = tb.getBatch(1); @@ -106,7 +64,7 @@ ElementSymbol x = new ElementSymbol("x"); //$NON-NLS-1$ x.setType(DataTypeManager.DefaultDataClasses.CLOB); List<ElementSymbol> schema = Arrays.asList(x); - TupleBuffer tb = new TupleBuffer(new FakeBatchManager(), "x", schema, LobManager.getLobIndexes(schema), 32); //$NON-NLS-1$ + TupleBuffer tb = BufferManagerFactory.getStandaloneBufferManager().createTupleBuffer(schema, "x", TupleSourceType.PROCESSOR); //$NON-NLS-1$ tb.setInlineLobs(false); ClobType c = new ClobType(new SerialClob(new char[0])); TupleBatch batch = new TupleBatch(1, new List[] {Arrays.asList(c)}); Added: trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestBitSetTree.java =================================================================== --- trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestBitSetTree.java (rev 0) +++ trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestBitSetTree.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -0,0 +1,60 @@ +/* + * JBoss, Home of Professional Open Source. + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301 USA. + */ + + package org.teiid.common.buffer.impl; + +import static org.junit.Assert.*; + +import java.util.BitSet; +import java.util.Random; + +import org.junit.Test; + +public class TestBitSetTree { + + @Test public void testBitsSet() { + BitSetTree bst = new BitSetTree(); + bst.set(1, true); + bst.set(100, true); + bst.set(10000, true); + bst.set(1000000, true); + assertEquals(4, bst.getBitsSet()); + } + + @Test public void testNextClearSet() { + BitSetTree bst = new BitSetTree(); + BitSet bst1 = new BitSet(); + Random r = new Random(1); + for (int i = 0; i < 1000; i++) { + int rand = r.nextInt() & BitSetTree.MAX_INDEX; + bst.set(rand, true); + bst1.set(rand, true); + } + + for (int i = 0; i < 10000; i++) { + int rand = r.nextInt() & BitSetTree.MAX_INDEX; + assertEquals(bst1.nextClearBit(rand), bst.nextClearBit(rand)); + assertEquals(bst1.nextSetBit(rand), bst.nextSetBit(rand)); + } + } + +} Property changes on: trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestBitSetTree.java ___________________________________________________________________ Added: svn:mime-type + text/plain Modified: trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestFileStorageManager.java =================================================================== --- trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestFileStorageManager.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestFileStorageManager.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -25,6 +25,7 @@ import static org.junit.Assert.*; import java.io.File; +import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import java.util.Arrays; @@ -59,7 +60,7 @@ assertEquals(0, sm.getUsedBufferSpace()); } - @Test(expected=TeiidComponentException.class) public void testMaxSpace() throws Exception { + @Test(expected=IOException.class) public void testMaxSpace() throws Exception { FileStorageManager sm = getStorageManager(null, null); sm.setMaxBufferSpace(1); String tsID = "0"; //$NON-NLS-1$ @@ -81,10 +82,11 @@ static Random r = new Random(); static void writeBytes(FileStore store) - throws TeiidComponentException { + throws IOException { byte[] bytes = new byte[2048]; r.nextBytes(bytes); - long start = store.write(bytes, 0, bytes.length); + long start = store.getLength(); + store.write(bytes, 0, bytes.length); byte[] bytesRead = new byte[2048]; store.readFully(start, bytesRead, 0, bytesRead.length); assertTrue(Arrays.equals(bytes, bytesRead)); Added: trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestFileStoreCache.java =================================================================== --- trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestFileStoreCache.java (rev 0) +++ trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestFileStoreCache.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -0,0 +1,77 @@ +/* + * JBoss, Home of Professional Open Source. + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301 USA. + */ + +package org.teiid.common.buffer.impl; + +import static org.junit.Assert.*; + +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.util.Arrays; +import java.util.List; + +import org.junit.Test; +import org.teiid.common.buffer.CacheEntry; +import org.teiid.common.buffer.Serializer; + +public class TestFileStoreCache { + + @Test public void testAddGetMultiBlock() throws Exception { + FileStoreCache fsc = new FileStoreCache(); + fsc.setStorageManager(new MemoryStorageManager()); + fsc.initialize(); + + CacheEntry ce = new CacheEntry(2l); + Serializer<Object> s = new Serializer<Object>() { + @Override + public Object deserialize(ObjectInputStream ois) + throws IOException, ClassNotFoundException { + return ois.readObject(); + } + + @Override + public Long getId() { + return 1l; + } + + @Override + public void serialize(Object obj, ObjectOutputStream oos) + throws IOException { + oos.writeObject(obj); + } + + @Override + public boolean useSoftCache() { + return false; + } + }; + fsc.createCacheGroup(s.getId()); + List<Object> cacheObject = Arrays.asList(new Object[10000]); + ce.setObject(cacheObject); + fsc.add(ce, s); + + ce = fsc.get(2l, s); + assertEquals(cacheObject, ce.getObject()); + } + +} Property changes on: trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestFileStoreCache.java ___________________________________________________________________ Added: svn:mime-type + text/plain Modified: trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestSizeUtility.java =================================================================== --- trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestSizeUtility.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestSizeUtility.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -34,7 +34,6 @@ import java.util.List; import org.junit.Test; -import org.teiid.common.buffer.TupleBatch; public class TestSizeUtility { @@ -43,7 +42,7 @@ } public void helpTestGetSize(Object obj, long expectedSize) { - long actualSize = new SizeUtility().getSize(obj, true, false); + long actualSize = new SizeUtility(null).getSize(obj, true, false); assertEquals("Got unexpected size: ", expectedSize, actualSize); //$NON-NLS-1$ } @@ -176,9 +175,7 @@ String[] types = {"string", "integer", "boolean", "double", "string", "integer"}; //$NON-NLS-1$//$NON-NLS-2$//$NON-NLS-3$//$NON-NLS-4$ //$NON-NLS-5$//$NON-NLS-6$ - TupleBatch tb = new TupleBatch(1, expected); - tb.setDataTypes(types); - long actualSize = new SizeUtility().getBatchSize(tb); + long actualSize = new SizeUtility(types).getBatchSize(Arrays.asList(expected)); assertEquals("Got unexpected size: ", 2667, actualSize); //$NON-NLS-1$ } Modified: trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestSplittableStorageManager.java =================================================================== --- trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestSplittableStorageManager.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/test/java/org/teiid/common/buffer/impl/TestSplittableStorageManager.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -43,10 +43,18 @@ TestFileStorageManager.writeBytes(store); assertEquals(2, msm.getCreated()); + + store.setLength(10000); + assertEquals(5, msm.getCreated()); + + store.setLength(100); + + assertEquals(4, msm.getRemoved()); + store.remove(); - assertEquals(2, msm.getRemoved()); + assertEquals(5, msm.getRemoved()); } @Test public void testTruncate() throws Exception { @@ -64,7 +72,7 @@ assertEquals(2, msm.getCreated()); - store.truncate(100); + store.setLength(100); assertEquals(1, msm.getRemoved()); Modified: trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestCachedResults.java =================================================================== --- trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestCachedResults.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestCachedResults.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -30,8 +30,9 @@ import org.teiid.cache.Cache; import org.teiid.cache.DefaultCache; import org.teiid.common.buffer.BufferManager; +import org.teiid.common.buffer.BufferManagerFactory; import org.teiid.common.buffer.TupleBuffer; -import org.teiid.common.buffer.TestTupleBuffer.FakeBatchManager; +import org.teiid.common.buffer.BufferManager.TupleSourceType; import org.teiid.core.types.DataTypeManager; import org.teiid.core.util.UnitTestUtil; import org.teiid.dqp.service.FakeBufferService; @@ -53,7 +54,7 @@ ElementSymbol x = new ElementSymbol("x"); //$NON-NLS-1$ x.setType(DataTypeManager.DefaultDataClasses.INTEGER); List<ElementSymbol> schema = Arrays.asList(x); - TupleBuffer tb = new TupleBuffer(new FakeBatchManager(), "x", schema, null, 4); //$NON-NLS-1$ + TupleBuffer tb = BufferManagerFactory.getStandaloneBufferManager().createTupleBuffer(schema, "x", TupleSourceType.PROCESSOR); //$NON-NLS-1$ tb.setForwardOnly(false); tb.addTuple(Arrays.asList(1)); Modified: trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java =================================================================== --- trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -311,7 +311,7 @@ reqMsg.setCursorType(ResultSet.TYPE_FORWARD_ONLY); DQPWorkContext.getWorkContext().getSession().setSessionId(sessionid); DQPWorkContext.getWorkContext().getSession().setUserName(userName); - ((BufferManagerImpl)core.getBufferManager()).setProcessorBatchSize(2); + ((BufferManagerImpl)core.getBufferManager()).setProcessorBatchSize(1); Future<ResultsMessage> message = core.executeRequest(reqMsg.getExecutionId(), reqMsg); ResultsMessage rm = message.get(500000, TimeUnit.MILLISECONDS); assertNull(rm.getException()); @@ -337,7 +337,7 @@ message = core.processCursorRequest(reqMsg.getExecutionId(), (j + 2) * rowsPerBatch + 1, rowsPerBatch); rm = message.get(5000, TimeUnit.MILLISECONDS); assertNull(rm.getException()); - assertEquals(rowsPerBatch, rm.getResults().length); + assertEquals(rowsPerBatch, rm.getResultsList().size()); } } @@ -351,11 +351,11 @@ reqMsg.setCursorType(ResultSet.TYPE_FORWARD_ONLY); DQPWorkContext.getWorkContext().getSession().setSessionId(sessionid); DQPWorkContext.getWorkContext().getSession().setUserName(userName); - ((BufferManagerImpl)core.getBufferManager()).setProcessorBatchSize(2); + ((BufferManagerImpl)core.getBufferManager()).setProcessorBatchSize(1); Future<ResultsMessage> message = core.executeRequest(reqMsg.getExecutionId(), reqMsg); ResultsMessage rm = message.get(500000, TimeUnit.MILLISECONDS); assertNull(rm.getException()); - assertEquals(8, rm.getResults().length); + assertEquals(8, rm.getResultsList().size()); RequestWorkItem item = core.getRequestWorkItem(DQPWorkContext.getWorkContext().getRequestID(reqMsg.getExecutionId())); assertEquals(100, item.resultsBuffer.getRowCount()); } Modified: trunk/engine/src/test/java/org/teiid/dqp/service/AutoGenDataService.java =================================================================== --- trunk/engine/src/test/java/org/teiid/dqp/service/AutoGenDataService.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/test/java/org/teiid/dqp/service/AutoGenDataService.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -95,7 +95,7 @@ results = new List[] {Arrays.asList(1)}; } - final AtomicResultsMessage msg = ConnectorWorkItem.createResultsMessage(results, projectedSymbols); + final AtomicResultsMessage msg = ConnectorWorkItem.createResultsMessage(results); msg.setFinalRow(rows); return new ConnectorWork() { Modified: trunk/engine/src/test/java/org/teiid/query/processor/TestTempTables.java =================================================================== --- trunk/engine/src/test/java/org/teiid/query/processor/TestTempTables.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/test/java/org/teiid/query/processor/TestTempTables.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -188,19 +188,19 @@ tempStore = gtsi.getTempTableStore(); metadata = new TempMetadataAdapter(RealMetadataFactory.example1Cached(), tempStore.getMetadataStore()); execute("create local temporary table x (e1 string, e2 integer)", new List[] {Arrays.asList(0)}); //$NON-NLS-1$ - for (int i = 0; i < 86; i++) { + for (int i = 0; i < 300; i++) { execute("insert into x (e2, e1) select e2, e1 from pm1.g1", new List[] {Arrays.asList(6)}); //$NON-NLS-1$ } setupTransaction(Connection.TRANSACTION_SERIALIZABLE); - execute("select count(*) from x", new List[] {Arrays.asList(516)}); - gtsi.updateMatViewRow("X", Arrays.asList(1), true); + execute("select count(e1) from x", new List[] {Arrays.asList(1500)}); + gtsi.updateMatViewRow("X", Arrays.asList(2), true); tc=null; //outside of the transaction we can see the row removed - execute("select count(*) from x", new List[] {Arrays.asList(515)}); + execute("select count(e1) from x", new List[] {Arrays.asList(1499)}); //back in the transaction we see the original state setupTransaction(Connection.TRANSACTION_SERIALIZABLE); - execute("select count(*) from x", new List[] {Arrays.asList(516)}); + execute("select count(e1) from x", new List[] {Arrays.asList(1500)}); synch.afterCompletion(Status.STATUS_COMMITTED); } Modified: trunk/engine/src/test/java/org/teiid/query/processor/TestVirtualDepJoin.java =================================================================== --- trunk/engine/src/test/java/org/teiid/query/processor/TestVirtualDepJoin.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/test/java/org/teiid/query/processor/TestVirtualDepJoin.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -33,9 +33,8 @@ import org.junit.Test; import org.teiid.common.buffer.BufferManager; +import org.teiid.common.buffer.BufferManagerFactory; import org.teiid.common.buffer.TupleBuffer; -import org.teiid.common.buffer.impl.BufferManagerImpl; -import org.teiid.common.buffer.impl.MemoryStorageManager; import org.teiid.core.TeiidComponentException; import org.teiid.core.TeiidException; import org.teiid.core.types.DataTypeManager; @@ -461,16 +460,8 @@ TestProcessor.examineResults((List[])expected.toArray(new List[expected.size()]), bufferMgr, id); } - private BufferManager createCustomBufferMgr(int batchSize) throws TeiidComponentException { - BufferManagerImpl bufferMgr = new BufferManagerImpl(); - bufferMgr.setConnectorBatchSize(batchSize); - bufferMgr.setProcessorBatchSize(batchSize); - bufferMgr.initialize(); - - // Add unmanaged memory storage manager - bufferMgr.setStorageManager(new MemoryStorageManager()); - - return bufferMgr; + private BufferManager createCustomBufferMgr(int batchSize) { + return BufferManagerFactory.getTestBufferManager(200000, batchSize); } public void helpTestVirtualDepJoin(boolean pushCriteria) throws Exception { Modified: trunk/engine/src/test/java/org/teiid/query/sql/symbol/TestElementSymbol.java =================================================================== --- trunk/engine/src/test/java/org/teiid/query/sql/symbol/TestElementSymbol.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/engine/src/test/java/org/teiid/query/sql/symbol/TestElementSymbol.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -127,22 +127,22 @@ @Test public void testEquals5() { ElementSymbol es = sampleElement1(); - helpEquals(es, (ElementSymbol)es.clone(), true); + helpEquals(es, es.clone(), true); } @Test public void testEquals6() { ElementSymbol es = sampleElement2(); - helpEquals(es, (ElementSymbol)es.clone(), true); + helpEquals(es, es.clone(), true); } @Test public void testEquals7() { ElementSymbol es = sampleElement3(); - helpEquals(es, (ElementSymbol)es.clone(), true); + helpEquals(es, es.clone(), true); } @Test public void testEquals8() { ElementSymbol es = sampleElement4(); - helpEquals(es, (ElementSymbol)es.clone(), true); + helpEquals(es, es.clone(), true); } // Compare fully-qualified to short versions @@ -190,7 +190,7 @@ @Test public void testCloneEquivalence(){ ElementSymbol s1 = sampleElement1(); - ElementSymbol s2 = (ElementSymbol)s1.clone(); + ElementSymbol s2 = s1.clone(); int equals = 0; UnitTestUtil.helpTestEquivalence(equals, s1, s2); } @@ -225,31 +225,31 @@ @Test public void testClone1() { ElementSymbol e1 = sampleElement1(); - ElementSymbol copy = (ElementSymbol) e1.clone(); + ElementSymbol copy = e1.clone(); helpEquals(e1, copy, true); } @Test public void testClone2() { ElementSymbol e1 = sampleElement2(); - ElementSymbol copy = (ElementSymbol) e1.clone(); + ElementSymbol copy = e1.clone(); helpEquals(e1, copy, true); } @Test public void testClone3() { ElementSymbol e1 = sampleElement3(); - ElementSymbol copy = (ElementSymbol) e1.clone(); + ElementSymbol copy = e1.clone(); helpEquals(e1, copy, true); } @Test public void testClone4() { ElementSymbol e1 = sampleElement4(); - ElementSymbol copy = (ElementSymbol) e1.clone(); + ElementSymbol copy = e1.clone(); helpEquals(e1, copy, true); } @Test public void testClone5() { ElementSymbol e1 = sampleElement1(); - ElementSymbol copy = (ElementSymbol) e1.clone(); + ElementSymbol copy = e1.clone(); helpEquals(e1, copy, true); // Change original, clone shouldn't change Modified: trunk/runtime/src/main/java/org/teiid/services/BufferServiceImpl.java =================================================================== --- trunk/runtime/src/main/java/org/teiid/services/BufferServiceImpl.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/runtime/src/main/java/org/teiid/services/BufferServiceImpl.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -33,6 +33,7 @@ import org.teiid.common.buffer.BufferManager; import org.teiid.common.buffer.impl.BufferManagerImpl; import org.teiid.common.buffer.impl.FileStorageManager; +import org.teiid.common.buffer.impl.FileStoreCache; import org.teiid.common.buffer.impl.MemoryStorageManager; import org.teiid.common.buffer.impl.SplittableStorageManager; import org.teiid.core.TeiidComponentException; @@ -105,10 +106,11 @@ fsm.setMaxBufferSpace(maxBufferSpace*MB); SplittableStorageManager ssm = new SplittableStorageManager(fsm); ssm.setMaxFileSize(maxFileSize); - ssm.initialize(); - this.bufferMgr.setStorageManager(ssm); + FileStoreCache fsc = new FileStoreCache(); + fsc.setStorageManager(ssm); + this.bufferMgr.setCache(fsc); } else { - this.bufferMgr.setStorageManager(new MemoryStorageManager()); + this.bufferMgr.setCache(new MemoryStorageManager()); } } catch(TeiidComponentException e) { @@ -129,7 +131,7 @@ } } - public BufferManager getBufferManager() { + public BufferManagerImpl getBufferManager() { return this.bufferMgr; } Modified: trunk/runtime/src/test/java/org/teiid/dqp/service/buffer/TestLocalBufferService.java =================================================================== --- trunk/runtime/src/test/java/org/teiid/dqp/service/buffer/TestLocalBufferService.java 2011-09-20 00:16:04 UTC (rev 3501) +++ trunk/runtime/src/test/java/org/teiid/dqp/service/buffer/TestLocalBufferService.java 2011-09-21 04:13:59 UTC (rev 3502) @@ -31,6 +31,7 @@ import org.teiid.common.buffer.BufferManager; import org.teiid.common.buffer.impl.BufferManagerImpl; import org.teiid.common.buffer.impl.FileStorageManager; +import org.teiid.common.buffer.impl.FileStoreCache; import org.teiid.common.buffer.impl.SplittableStorageManager; import org.teiid.core.types.DataTypeManager; import org.teiid.core.util.UnitTestUtil; @@ -53,8 +54,8 @@ assertTrue("does not end with one", svc.getBufferDirectory().getParent().endsWith("1")); //$NON-NLS-1$ //$NON-NLS-2$ assertTrue(svc.isUseDisk()); - BufferManagerImpl mgr = (BufferManagerImpl) svc.getBufferManager(); - SplittableStorageManager ssm = (SplittableStorageManager)mgr.getStorageManager(); + BufferManagerImpl mgr = svc.getBufferManager(); + SplittableStorageManager ssm = (SplittableStorageManager)((FileStoreCache)mgr.getCache()).getStorageManager(); assertTrue(((FileStorageManager)ssm.getStorageManager()).getDirectory().endsWith(svc.getBufferDirectory().getName())); }