[teiid-commits] teiid SVN: r1705 - in trunk: engine/src/main/java/com/metamatrix/common/buffer and 17 other directories.

Author: shawkins Date: 2009-12-24 00:21:00 -0500 (Thu, 24 Dec 2009) New Revision: 1705 Removed: trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManagerPropertyNames.java trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BufferConfig.java trunk/engine/src/main/java/com/metamatrix/common/buffer/storage/file/ trunk/engine/src/main/java/com/metamatrix/connector/metadata/ trunk/engine/src/main/java/org/teiid/dqp/internal/process/CacheResults.java Modified: trunk/build/kit-runtime/deploy.properties trunk/engine/src/main/java/com/metamatrix/common/buffer/BlockedException.java trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManager.java trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BufferManagerImpl.java trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/FileStorageManager.java trunk/engine/src/main/java/com/metamatrix/dqp/message/AtomicRequestMessage.java trunk/engine/src/main/java/com/metamatrix/query/optimizer/relational/rules/NewCalculateCostUtil.java trunk/engine/src/main/java/com/metamatrix/query/processor/TempTableDataManager.java trunk/engine/src/main/java/com/metamatrix/query/processor/relational/DependentCriteriaProcessor.java trunk/engine/src/main/java/com/metamatrix/query/processor/relational/GroupingNode.java trunk/engine/src/main/java/com/metamatrix/query/processor/relational/NestedLoopJoinStrategy.java trunk/engine/src/main/java/com/metamatrix/query/processor/relational/SortUtility.java trunk/engine/src/main/java/com/metamatrix/query/util/CommandContext.java trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/impl/ExecutionContextImpl.java trunk/engine/src/test/java/com/metamatrix/common/buffer/BufferManagerFactory.java trunk/engine/src/test/java/com/metamatrix/common/buffer/storage/file/TestFileStorageManager.java trunk/engine/src/test/java/com/metamatrix/query/processor/TestProcessor.java trunk/engine/src/test/java/com/metamatrix/query/processor/TestVirtualDepJoin.java trunk/engine/src/test/java/com/metamatrix/query/processor/relational/NodeTestUtil.java trunk/engine/src/test/java/com/metamatrix/query/processor/relational/TestGroupingNode.java trunk/engine/src/test/java/com/metamatrix/query/processor/relational/TestJoinNode.java trunk/runtime/src/main/java/com/metamatrix/dqp/embedded/admin/DQPRuntimeStateAdminImpl.java trunk/runtime/src/main/java/com/metamatrix/dqp/embedded/services/EmbeddedBufferService.java trunk/runtime/src/test/java/com/metamatrix/dqp/service/buffer/TestLocalBufferService.java Log: TEIID-913 intermediate commit in removing old implementation - chaning the defaults for batch sizes and adding a new property to control max processing batches. Modified: trunk/build/kit-runtime/deploy.properties =================================================================== --- trunk/build/kit-runtime/deploy.properties 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/build/kit-runtime/deploy.properties 2009-12-24 05:21:00 UTC (rev 1705) @@ -47,11 +47,11 @@ # BufferManager Settings # -#The max size of a batch sent between connector and query service. Should be even multiple of processorBatchSize. (default 2000) -dqp.buffer.connectorBatchSize=2000 +#The max size of a batch sent between connector and query service. Should be even multiple of processorBatchSize. (default 2048) +dqp.buffer.connectorBatchSize=2048 -#The max size of a batch sent internally within the query processor. Should be <= the connectorBatchSize. (default 2000) -dqp.buffer.processorBatchSize=2000 +#The max size of a batch sent internally within the query processor. Should be <= the connectorBatchSize. (default 1024) +dqp.buffer.processorBatchSize=1024 #Defines whether to use disk buffering or not. (default true) dqp.buffer.useDisk=true Modified: trunk/engine/src/main/java/com/metamatrix/common/buffer/BlockedException.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/common/buffer/BlockedException.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/common/buffer/BlockedException.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -31,7 +31,7 @@ */ public class BlockedException extends MetaMatrixComponentException { - public static final BlockedException INSTANCE = new BlockedException(""); //$NON-NLS-1$ + public static final BlockedException INSTANCE = new BlockedException(); /** * No-arg costructor required by Externalizable semantics @@ -40,44 +40,4 @@ super(); } - /** - * Construct an instance with the message specified. - * - * @param message A message describing the exception - */ - public BlockedException( String message ) { - super( message ); - } - - /** - * Construct an instance with the message and error code specified. - * - * @param message A message describing the exception - * @param code The error code - */ - public BlockedException( String code, String message ) { - super( code, message ); - } - - /** - * Construct an instance from a message and an exception to chain to this one. - * - * @param message A message describing the exception - * @param e An exception to nest within this one - */ - public BlockedException( Throwable e, String message ) { - super( e, message ); - } - - /** - * Construct an instance from a message and a code and an exception to - * chain to this one. - * - * @param e An exception to nest within this one - * @param message A message describing the exception - * @param code A code denoting the exception - */ - public BlockedException( Throwable e, String code, String message ) { - super( e, code, message ); - } } Modified: trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManager.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManager.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManager.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -48,6 +48,48 @@ FINAL } + /** + * Optional property - the max size of a batch sent between connector and query service. + * Making batches larger reduces communication overhead between connector and query service + * but increases the granularity of memory management on those batches. This value should + * be a positive integer and defaults to 1000. + */ + public static final String CONNECTOR_BATCH_SIZE = "metamatrix.buffer.connectorBatchSize"; //$NON-NLS-1$ + /** + * Optional property - the max size of a batch sent internally within the query processor. + * In general, these batches should be smaller than the connector batch size as there are + * no communication costs with these batches. Smaller batches typically allow a user to + * get their first results quicker and allow fine-grained buffer management on intermediate + * results. This value should be a positive integer and defaults to 100. + */ + public static final String PROCESSOR_BATCH_SIZE = "metamatrix.buffer.processorBatchSize"; //$NON-NLS-1$ + /** + * Optional property - this value specifies the location to store temporary buffers to + * large to fit in memory. Temporary buffer files will be created and destroyed in this + * directory. This value should be a string specifying an absolute directory path. + */ + public static final String BUFFER_STORAGE_DIRECTORY = "metamatrix.buffer.storageDirectory"; //$NON-NLS-1$ + /** + * Optional property - this values specifies how many open file descriptors should be cached + * in the storage directory. Increasing this value in heavy load may improve performance + * but will use more file descriptors, which are a limited system resource. The default + * is 10. + */ + public static final String MAX_OPEN_FILES = "metamatrix.buffer.maxOpenFiles"; //$NON-NLS-1$ + /** + * Optional property - this values specifies the maximum size in MegaBytes that a buffer file can reach. + * The default is 2048 MB (i.e. 2GB). + */ + public static final String MAX_FILE_SIZE = "metamatrix.buffer.maxFileSize"; //$NON-NLS-1$ + /** + * Optional property - the max number of batches to process at once in algorithms such as sorting. + */ + public static final String MAX_PROCESSING_BATCHES = "metamatrix.buffer.maxProcessingBatches"; //$NON-NLS-1$ + + public static int DEFAULT_CONNECTOR_BATCH_SIZE = 2048; + public static int DEFAULT_PROCESSOR_BATCH_SIZE = 1024; + public static int DEFAULT_MAX_PROCESSING_BATCHES = 16; + /** * Get the batch size to use during query processing. * @return Batch size (# of rows) @@ -73,4 +115,6 @@ void removeTupleBuffers(String groupName) throws MetaMatrixComponentException; + int getMaxProcessingBatches(); + } Deleted: trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManagerPropertyNames.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManagerPropertyNames.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManagerPropertyNames.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -1,158 +0,0 @@ -/* - * 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 com.metamatrix.common.buffer; - -/** - * This class holds constants for all the buffer manager properties. - */ -public final class BufferManagerPropertyNames { - - /** - * Optional property - the amount of memory (in megabytes) that buffer management should use. - * This value should be a positive integer (less than the the max heap size) and defaults to - * 128. - */ - public static final String MEMORY_AVAILABLE = "metamatrix.buffer.memoryAvailable"; //$NON-NLS-1$ - - /** - * Optional property - the percent of buffer management memory that a particular session - * can use. This property can be used to prevent a single user from consuming too many - * resources. This value should be in the range [1..100] and defaults to 50. - */ - public static final String SESSION_USE_PERCENTAGE = "metamatrix.buffer.sessionUsePercentage"; //$NON-NLS-1$ - - /** - * Optional property - the percent of buffer management memory that serves as a threshold - * for active memory management. Below the threshold, no active memory management occurs - * and no data is moved between memory and persistent storage in the background. Above - * the threshold, a background thread attempts to clean the in-memory buffers using an LRU - * algorithm. This value should be in the range [1..100] and defaults to 75. - */ - public static final String ACTIVE_MEMORY_THRESHOLD = "metamatrix.buffer.activeMemoryThreshold"; //$NON-NLS-1$ - - /** - * Optional property - the period between checking whether active memory clean up - * should occur, in milliseconds. This value should be a millisecond value and defaults - * to 500 ms. A value of 0 indicates that no active management should occur. - */ - public static final String MANAGEMENT_INTERVAL = "metamatrix.buffer.managementInterval"; //$NON-NLS-1$ - - /** - * Optional property - the max size of a batch sent between connector and query service. - * Making batches larger reduces communication overhead between connector and query service - * but increases the granularity of memory management on those batches. This value should - * be a positive integer and defaults to 1000. - */ - public static final String CONNECTOR_BATCH_SIZE = "metamatrix.buffer.connectorBatchSize"; //$NON-NLS-1$ - - /** - * Optional property - the max size of a batch sent internally within the query processor. - * In general, these batches should be smaller than the connector batch size as there are - * no communication costs with these batches. Smaller batches typically allow a user to - * get their first results quicker and allow fine-grained buffer management on intermediate - * results. This value should be a positive integer and defaults to 100. - */ - public static final String PROCESSOR_BATCH_SIZE = "metamatrix.buffer.processorBatchSize"; //$NON-NLS-1$ - - /** - * Optional property - this value specifies the location to store temporary buffers to - * large to fit in memory. Temporary buffer files will be created and destroyed in this - * directory. This value should be a string specifying an absolute directory path. - */ - public static final String BUFFER_STORAGE_DIRECTORY = "metamatrix.buffer.storageDirectory"; //$NON-NLS-1$ - - /** - * Optional property - this values specifies how many open file descriptors should be cached - * in the storage directory. Increasing this value in heavy load may improve performance - * but will use more file descriptors, which are a limited system resource. The default - * is 10. - */ - public static final String MAX_OPEN_FILES = "metamatrix.buffer.maxOpenFiles"; //$NON-NLS-1$ - /** - * Optional property - this values specifies the maximum size in MegaBytes that a buffer file can reach. - * The default is 2048 MB (i.e. 2GB). - */ - public static final String MAX_FILE_SIZE = "metamatrix.buffer.maxFileSize"; //$NON-NLS-1$ - - /** - * Optional property - this value specifies how often the buffer statistics should be - * collected and logged. This is primarily useful during debugging and defaults to a value - * of 0, which indicates no stats logging. This value should be either 0 to indicate no - * logging or a positive integer indicating the period in milliseconds between logging. - */ - public static final String LOG_STATS_INTERVAL = "metamatrix.buffer.logStatsInterval"; //$NON-NLS-1$ - - /** - * The environment property name for the class that is to be used for the MetadataConnectionFactory implementation. - * This property is required (there is no default). - */ - public static final String CONNECTION_FACTORY = "metamatrix.buffer.connection.Factory"; //$NON-NLS-1$ - - /** - * The environment property name for the class of the driver. - * This property is optional. - */ - public static final String CONNECTION_DRIVER = "metamatrix.buffer.connection.Driver"; //$NON-NLS-1$ - - /** - * The environment property name for the protocol for connecting to the metadata store. - * This property is optional. - */ - public static final String CONNECTION_PROTOCOL = "metamatrix.buffer.connection.Protocol"; //$NON-NLS-1$ - - /** - * The environment property name for the name of the metadata store database. - * This property is optional. - */ - public static final String CONNECTION_DATABASE = "metamatrix.buffer.connection.Database"; //$NON-NLS-1$ - - /** - * The environment property name for the username that is to be used for connecting to the metadata store. - * This property is optional. - */ - public static final String CONNECTION_USERNAME = "metamatrix.buffer.connection.User"; //$NON-NLS-1$ - - /** - * The environment property name for the password that is to be used for connecting to the metadata store. - * This property is optional. - */ - public static final String CONNECTION_PASSWORD = "metamatrix.buffer.connection.Password"; //$NON-NLS-1$ - - /** - * The environment property name for the maximum number of milliseconds that a metadata connection - * may remain unused before it becomes a candidate for garbage collection. - * This property is optional. - */ - public static final String CONNECTION_POOL_MAXIMUM_AGE = "metamatrix.buffer.connection.MaximumAge"; //$NON-NLS-1$ - - /** - * The environment property name for the maximum number of concurrent users of a single metadata connection. - * This property is optional. - */ - public static final String CONNECTION_POOL_MAXIMUM_CONCURRENT_USERS = "metamatrix.buffer.connection.MaximumConcurrentReaders"; //$NON-NLS-1$ - - - // Can't construct - private BufferManagerPropertyNames() { - } -} Deleted: trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BufferConfig.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BufferConfig.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BufferConfig.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -1,232 +0,0 @@ -/* - * 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 com.metamatrix.common.buffer.impl; - -import java.util.Properties; - -import com.metamatrix.common.buffer.BufferManagerPropertyNames; - -/** - * Encapsulates all configuration information for the BufferManagerImpl, - * including both properties that are set and some that are derived. - */ -public class BufferConfig { - - public static int DEFAULT_CONNECTOR_BATCH_SIZE = 2000; - public static int DEFAULT_PROCESSOR_BATCH_SIZE = 2000; - - // Configuration - private long totalAvailableMemory = 100000000; - private int groupUsePercentage = 80; - private int activeMemoryThreshold = 75; - private int managementInterval = 500; - private int connectorBatchSize = DEFAULT_CONNECTOR_BATCH_SIZE; - private int processorBatchSize = DEFAULT_PROCESSOR_BATCH_SIZE; - private String bufferStorageDirectory = "../buffer"; //$NON-NLS-1$ - private int logStatInterval = 0; - - // Derived state - private long availableSessionLevel = 0; - private long activeMemoryLevel = 0; - - /** - * Constructor for BufferConfig - use all defaults - */ - public BufferConfig() { - computeDerived(); - } - - /** - * Constructor for BufferConfig - set from properties. - * @param props Properties as defined in - * {@link com.metamatrix.common.buffer.BufferManagerPropertyNames}. - */ - public BufferConfig(Properties props) { - // Read totalAvailableMemory - try { - String propStr = props.getProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE); - if(propStr != null) { - totalAvailableMemory = Integer.parseInt(propStr) * 1000000L; - } - } catch(NumberFormatException e) { - // use default - } - - // Read groupUsePercentage - try { - String propStr = props.getProperty(BufferManagerPropertyNames.SESSION_USE_PERCENTAGE); - if(propStr != null) { - groupUsePercentage = Integer.parseInt(propStr); - } - } catch(NumberFormatException e) { - // use default - } - - // Read activeMemoryThreshold - try { - String propStr = props.getProperty(BufferManagerPropertyNames.ACTIVE_MEMORY_THRESHOLD); - if(propStr != null) { - activeMemoryThreshold = Integer.parseInt(propStr); - } - } catch(NumberFormatException e) { - // use default - } - - // Read managementInterval - try { - String propStr = props.getProperty(BufferManagerPropertyNames.MANAGEMENT_INTERVAL); - if(propStr != null) { - managementInterval = Integer.parseInt(propStr); - } - } catch(NumberFormatException e) { - // use default - } - - // Read connectorBatchSize - try { - String propStr = props.getProperty(BufferManagerPropertyNames.CONNECTOR_BATCH_SIZE); - if(propStr != null) { - connectorBatchSize = Integer.parseInt(propStr); - } - } catch(NumberFormatException e) { - // use default - } - - // Read processorBatchSize - try { - String propStr = props.getProperty(BufferManagerPropertyNames.PROCESSOR_BATCH_SIZE); - if(propStr != null) { - processorBatchSize = Integer.parseInt(propStr); - } - } catch(NumberFormatException e) { - // use default - } - - // Read bufferStorageDirectory - try { - String propStr = props.getProperty(BufferManagerPropertyNames.BUFFER_STORAGE_DIRECTORY); - if(propStr != null) { - bufferStorageDirectory = propStr; - } - } catch(NumberFormatException e) { - // use default - } - - // Read logStatInterval - try { - String propStr = props.getProperty(BufferManagerPropertyNames.LOG_STATS_INTERVAL); - if(propStr != null) { - logStatInterval = Integer.parseInt(propStr); - } - } catch(NumberFormatException e) { - // use default - } - - computeDerived(); - } - - // Direct state management - - public long getTotalAvailableMemory() { - return this.totalAvailableMemory; - } - - public void setTotalAvailableMemory(long totalAvailableMemory) { - this.totalAvailableMemory = totalAvailableMemory; - computeDerived(); - } - - public int getGroupUsePercentage() { - return this.groupUsePercentage; - } - - public void setGroupUsePercentage(int groupUsePercentage) { - this.groupUsePercentage = groupUsePercentage; - computeDerived(); - } - - public int getActiveMemoryThreshold() { - return this.activeMemoryThreshold; - } - - public void setActiveMemoryThreshold(int activeMemoryThreshold) { - this.activeMemoryThreshold = activeMemoryThreshold; - computeDerived(); - } - - public int getConnectorBatchSize() { - return this.connectorBatchSize; - } - - public void setConnectorBatchSize(int connectorBatchSize) { - this.connectorBatchSize = connectorBatchSize; - } - - public int getProcessorBatchSize() { - return this.processorBatchSize; - } - - public void setProcessorBatchSize(int processorBatchSize) { - this.processorBatchSize = processorBatchSize; - } - - public int getManagementInterval() { - return this.managementInterval; - } - - public void setManagementInterval(int managementInterval) { - this.managementInterval = managementInterval; - } - - public String getBufferStorageDirectory() { - return this.bufferStorageDirectory; - } - - public void setBufferStorageDirectory(String bufferStorageDirectory) { - this.bufferStorageDirectory = bufferStorageDirectory; - } - - public int getLogStatInterval() { - return this.logStatInterval; - } - - public void setStatUpdateInterval(int logStatInterval) { - this.logStatInterval = logStatInterval; - } - - // Derived state management - - private void computeDerived() { - this.availableSessionLevel = (int) (this.totalAvailableMemory * (this.groupUsePercentage / 100.0)); - this.activeMemoryLevel = (int) (this.totalAvailableMemory * (this.activeMemoryThreshold / 100.0)); - } - - public long getMaxAvailableSession() { - return this.availableSessionLevel; - } - - public long getActiveMemoryLevel() { - return this.activeMemoryLevel; - } - -} Modified: trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BufferManagerImpl.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BufferManagerImpl.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BufferManagerImpl.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -43,6 +43,7 @@ import com.metamatrix.common.buffer.TupleSourceNotFoundException; import com.metamatrix.common.log.LogManager; import com.metamatrix.common.types.DataTypeManager; +import com.metamatrix.common.util.PropertiesUtils; import com.metamatrix.core.log.MessageLevel; import com.metamatrix.core.util.Assertion; import com.metamatrix.dqp.util.LogConstants; @@ -55,31 +56,34 @@ */ public class BufferManagerImpl implements BufferManager, StorageManager { - // Initialized stuff private String lookup; - private BufferConfig config; - + + // Configuration + private int connectorBatchSize = BufferManager.DEFAULT_CONNECTOR_BATCH_SIZE; + private int processorBatchSize = BufferManager.DEFAULT_PROCESSOR_BATCH_SIZE; + private int maxProcessingBatches = BufferManager.DEFAULT_MAX_PROCESSING_BATCHES; + private Map<TupleSourceID, TupleBuffer> tupleSourceMap = new ConcurrentHashMap<TupleSourceID, TupleBuffer>(); private Map<String, Set<TupleSourceID>> groupInfos = new HashMap<String, Set<TupleSourceID>>(); private StorageManager diskMgr; private AtomicLong currentTuple = new AtomicLong(0); + + public int getMaxProcessingBatches() { + return maxProcessingBatches; + } + + public void setMaxProcessingBatches(int maxProcessingBatches) { + this.maxProcessingBatches = maxProcessingBatches; + } /** - * Get the configuration of the buffer manager - * @return Configuration - */ - public BufferConfig getConfig() { - return this.config; - } - - /** * Get processor batch size * @return Number of rows in a processor batch */ public int getProcessorBatchSize() { - return config.getProcessorBatchSize(); + return this.processorBatchSize; } /** @@ -87,9 +91,17 @@ * @return Number of rows in a connector batch */ public int getConnectorBatchSize() { - return config.getConnectorBatchSize(); + return this.connectorBatchSize; } + + public void setConnectorBatchSize(int connectorBatchSize) { + this.connectorBatchSize = connectorBatchSize; + } + public void setProcessorBatchSize(int processorBatchSize) { + this.processorBatchSize = processorBatchSize; + } + /** * Add a storage manager to this buffer manager, order is unimportant * @param storageManager Storage manager to add @@ -100,6 +112,10 @@ this.diskMgr = storageManager; } + public StorageManager getStorageManager() { + return diskMgr; + } + @Override public TupleBuffer createTupleBuffer(List elements, String groupName, TupleSourceType tupleSourceType) @@ -224,8 +240,8 @@ @Override public void initialize(Properties props) throws MetaMatrixComponentException { - this.config = new BufferConfig(props); this.lookup = "local"; //$NON-NLS-1$ + PropertiesUtils.setBeanProperties(this, props, "metamatrix.buffer"); //$NON-NLS-1$ } @Override Modified: trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/FileStorageManager.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/FileStorageManager.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/FileStorageManager.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -40,7 +40,7 @@ import java.util.Properties; import com.metamatrix.api.exception.MetaMatrixComponentException; -import com.metamatrix.common.buffer.BufferManagerPropertyNames; +import com.metamatrix.common.buffer.BufferManager; import com.metamatrix.common.buffer.StorageManager; import com.metamatrix.common.buffer.TupleBatch; import com.metamatrix.common.buffer.TupleSourceID; @@ -90,12 +90,12 @@ /** * Initialize with properties * @param props Initialization properties - * @see com.metamatrix.common.buffer.BufferManagerPropertyNames#BUFFER_STORAGE_DIRECTORY - * @see com.metamatrix.common.buffer.BufferManagerPropertyNames#MAX_OPEN_FILES - * @see com.metamatrix.common.buffer.BufferManagerPropertyNames#MAX_FILE_SIZE + * @see com.metamatrix.common.buffer.BufferManager#BUFFER_STORAGE_DIRECTORY + * @see com.metamatrix.common.buffer.BufferManager#MAX_OPEN_FILES + * @see com.metamatrix.common.buffer.BufferManager#MAX_FILE_SIZE */ public void initialize(Properties props) throws MetaMatrixComponentException { - this.directory = props.getProperty(BufferManagerPropertyNames.BUFFER_STORAGE_DIRECTORY); + this.directory = props.getProperty(BufferManager.BUFFER_STORAGE_DIRECTORY); if(this.directory == null) { throw new MetaMatrixComponentException(QueryExecPlugin.Util.getString("FileStoreageManager.no_directory")); //$NON-NLS-1$ } @@ -111,11 +111,15 @@ } // Set up max number of open file descriptors - maxOpenFiles = PropertiesUtils.getIntProperty(props, BufferManagerPropertyNames.MAX_OPEN_FILES, 10); + maxOpenFiles = PropertiesUtils.getIntProperty(props, BufferManager.MAX_OPEN_FILES, 10); // Set the max file size - maxFileSize = PropertiesUtils.getIntProperty(props, BufferManagerPropertyNames.MAX_FILE_SIZE, 2048) * 1024L * 1024L; // Multiply by 1MB + maxFileSize = PropertiesUtils.getIntProperty(props, BufferManager.MAX_FILE_SIZE, 2048) * 1024L * 1024L; // Multiply by 1MB } + + public String getDirectory() { + return directory; + } /** * Look up tuple source info and possibly create. First the file map is used to find an Modified: trunk/engine/src/main/java/com/metamatrix/dqp/message/AtomicRequestMessage.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/dqp/message/AtomicRequestMessage.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/dqp/message/AtomicRequestMessage.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -31,7 +31,7 @@ import org.teiid.connector.xa.api.TransactionContext; import org.teiid.dqp.internal.process.DQPWorkContext; -import com.metamatrix.common.buffer.impl.BufferConfig; +import com.metamatrix.common.buffer.BufferManager; import com.metamatrix.dqp.internal.datamgr.ConnectorID; import com.metamatrix.query.sql.lang.Command; @@ -74,7 +74,7 @@ private Command command; // results fetch size - private int fetchSize = BufferConfig.DEFAULT_CONNECTOR_BATCH_SIZE; + private int fetchSize = BufferManager.DEFAULT_CONNECTOR_BATCH_SIZE; // The time when the command was created by the client private Date submittedTimestamp; Modified: trunk/engine/src/main/java/com/metamatrix/query/optimizer/relational/rules/NewCalculateCostUtil.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/query/optimizer/relational/rules/NewCalculateCostUtil.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/query/optimizer/relational/rules/NewCalculateCostUtil.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -36,7 +36,7 @@ import com.metamatrix.api.exception.MetaMatrixComponentException; import com.metamatrix.api.exception.query.QueryMetadataException; -import com.metamatrix.common.buffer.impl.BufferConfig; +import com.metamatrix.common.buffer.BufferManager; import com.metamatrix.common.log.LogManager; import com.metamatrix.common.types.DataTypeManager; import com.metamatrix.query.QueryPlugin; @@ -904,7 +904,7 @@ float numberComparisons = merge?(leftChildCardinality + rightChildCardinality):(leftChildCardinality * rightChildCardinality); - float connectorBatchSize = BufferConfig.DEFAULT_CONNECTOR_BATCH_SIZE; + float connectorBatchSize = BufferManager.DEFAULT_CONNECTOR_BATCH_SIZE; if(context != null) { connectorBatchSize = context.getConnectorBatchSize(); } @@ -965,8 +965,8 @@ return UNKNOWN_VALUE; } - float connectorBatchSize = BufferConfig.DEFAULT_CONNECTOR_BATCH_SIZE; - float processorBatchSize = BufferConfig.DEFAULT_PROCESSOR_BATCH_SIZE; + float connectorBatchSize = BufferManager.DEFAULT_CONNECTOR_BATCH_SIZE; + float processorBatchSize = BufferManager.DEFAULT_PROCESSOR_BATCH_SIZE; if(context != null) { connectorBatchSize = context.getConnectorBatchSize(); processorBatchSize = context.getProcessorBatchSize(); Modified: trunk/engine/src/main/java/com/metamatrix/query/processor/TempTableDataManager.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/query/processor/TempTableDataManager.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/query/processor/TempTableDataManager.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -26,7 +26,6 @@ import com.metamatrix.api.exception.MetaMatrixProcessingException; import com.metamatrix.common.buffer.BlockedException; import com.metamatrix.common.buffer.TupleSource; -import com.metamatrix.common.buffer.TupleSourceID; import com.metamatrix.query.sql.lang.Command; import com.metamatrix.query.tempdata.TempTableStore; import com.metamatrix.query.util.CommandContext; @@ -62,7 +61,6 @@ * If a temp group is <i>not</i> being selected from, then this request will be * passed through to the underlying ProcessorDataManager. * @throws MetaMatrixProcessingException - * @see com.metamatrix.query.processor.ProcessorDataManager#registerRequest(Object, Command, String, String, TupleSourceID) */ public TupleSource registerRequest( Object processorID, Modified: trunk/engine/src/main/java/com/metamatrix/query/processor/relational/DependentCriteriaProcessor.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/query/processor/relational/DependentCriteriaProcessor.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/query/processor/relational/DependentCriteriaProcessor.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -97,7 +97,7 @@ } } - public void close() throws MetaMatrixComponentException { + public void close() { if (dvs != null) { sortUtility = null; dvs.getTupleBuffer().remove(); Modified: trunk/engine/src/main/java/com/metamatrix/query/processor/relational/GroupingNode.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/query/processor/relational/GroupingNode.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/query/processor/relational/GroupingNode.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -194,7 +194,7 @@ functions[i] = new Max(); } - if(aggSymbol.isDistinct()) { + if(aggSymbol.isDistinct() && !function.equals(ReservedWords.MIN) && !function.equals(ReservedWords.MAX)) { functions[i] = new DuplicateFilter(functions[i], getBufferManager(), getConnectionID()); } Modified: trunk/engine/src/main/java/com/metamatrix/query/processor/relational/NestedLoopJoinStrategy.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/query/processor/relational/NestedLoopJoinStrategy.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/query/processor/relational/NestedLoopJoinStrategy.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -24,9 +24,6 @@ import java.util.List; -import com.metamatrix.api.exception.MetaMatrixComponentException; -import com.metamatrix.api.exception.MetaMatrixProcessingException; - /** * Nested loop is currently implemented as a degenerate case of merge join. * Modified: trunk/engine/src/main/java/com/metamatrix/query/processor/relational/SortUtility.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/query/processor/relational/SortUtility.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/query/processor/relational/SortUtility.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -24,8 +24,8 @@ import java.util.ArrayList; import java.util.Collections; -import java.util.Iterator; import java.util.List; +import java.util.ListIterator; import com.metamatrix.api.exception.MetaMatrixComponentException; import com.metamatrix.api.exception.MetaMatrixProcessingException; @@ -73,15 +73,15 @@ //constructor state private TupleSource sourceID; private Mode mode; - protected BufferManager bufferManager; + private BufferManager bufferManager; private String groupName; - protected List schema; + private List<SingleElementSymbol> schema; private ListNestedSortComparator comparator; private TupleBuffer output; private boolean doneReading; private int phase = INITIAL_SORT; - protected List<TupleBuffer> activeTupleBuffers = new ArrayList<TupleBuffer>(); + private List<TupleBuffer> activeTupleBuffers = new ArrayList<TupleBuffer>(); private int masterSortIndex; // Phase constants for readability @@ -100,7 +100,7 @@ if (mode != Mode.SORT) { if (sortElements != null && sortElements.size() < schema.size()) { sortElements = new ArrayList(sortElements); - List toAdd = new ArrayList(schema); + List<SingleElementSymbol> toAdd = new ArrayList<SingleElementSymbol>(schema); toAdd.removeAll(sortElements); sortElements.addAll(toAdd); sortTypes = new ArrayList<Boolean>(sortTypes); @@ -113,13 +113,11 @@ int[] cols = new int[sortElements.size()]; - Iterator iter = sortElements.iterator(); - - for (int i = 0; i < cols.length; i++) { - SingleElementSymbol elem = (SingleElementSymbol)iter.next(); + for (ListIterator<SingleElementSymbol> iter = sortElements.listIterator(); iter.hasNext();) { + SingleElementSymbol elem = iter.next(); - cols[i] = schema.indexOf(elem); - Assertion.assertTrue(cols[i] != -1); + cols[iter.previousIndex()] = schema.indexOf(elem); + Assertion.assertTrue(cols[iter.previousIndex()] != -1); } this.comparator = new ListNestedSortComparator(cols, sortTypes); this.comparator.setDistinctIndex(distinctIndex); @@ -150,12 +148,13 @@ } /** - * creates sort sublists stored in tuplebuffers + * creates sorted sublists stored in tuplebuffers */ protected void initialSort() throws MetaMatrixComponentException, MetaMatrixProcessingException { while(!doneReading) { - List<List<Object>> workingTuples = new ArrayList<List<Object>>(); - while(!doneReading) { //TODO: limit rows + List<List<?>> workingTuples = new ArrayList<List<?>>(); + int maxRows = bufferManager.getMaxProcessingBatches() * bufferManager.getProcessorBatchSize(); + while(!doneReading && workingTuples.size() < maxRows) { try { List<?> tuple = sourceID.nextTuple(); @@ -183,7 +182,7 @@ //perform a stable sort Collections.sort(workingTuples, comparator); } - for (List<Object> list : workingTuples) { + for (List<?> list : workingTuples) { activeID.addTuple(list); } } @@ -195,7 +194,7 @@ this.phase = MERGE; } - protected void addTuple(List workingTuples, List tuple) { + protected void addTuple(List<List<?>> workingTuples, List<?> tuple) { if (this.mode == Mode.SORT) { workingTuples.add(tuple); return; @@ -214,7 +213,8 @@ TupleBuffer merged = createTupleBuffer(); int sortedIndex = 0; - for(; sortedIndex<activeTupleBuffers.size(); sortedIndex++) { //TODO: limit activeTupleIDs + int maxSortIndex = Math.min(this.bufferManager.getMaxProcessingBatches() * 2, activeTupleBuffers.size()); + for(; sortedIndex<maxSortIndex; sortedIndex++) { TupleBuffer activeID = activeTupleBuffers.get(sortedIndex); SortedSublist sortedSublist = new SortedSublist(); sortedSublist.its = activeID.createIndexedTupleSource(); @@ -228,7 +228,7 @@ if (!sortedSublist.duplicate) { merged.addTuple(sortedSublist.tuple); if (this.output != null && sortedSublist.index != masterSortIndex && sortedIndex > masterSortIndex) { - this.output.addTuple(sortedSublist.tuple); + this.output.addTuple(sortedSublist.tuple); //a new distinct row } } addWorkingTuple(workingTuples, sortedSublist); Modified: trunk/engine/src/main/java/com/metamatrix/query/util/CommandContext.java =================================================================== --- trunk/engine/src/main/java/com/metamatrix/query/util/CommandContext.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/com/metamatrix/query/util/CommandContext.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -31,7 +31,7 @@ import com.metamatrix.api.exception.MetaMatrixComponentException; import com.metamatrix.api.exception.query.QueryProcessingException; -import com.metamatrix.common.buffer.impl.BufferConfig; +import com.metamatrix.common.buffer.BufferManager; import com.metamatrix.core.util.ArgCheck; import com.metamatrix.query.QueryPlugin; import com.metamatrix.query.eval.SecurityFunctionEvaluator; @@ -57,9 +57,9 @@ /** Identify a group of related commands, which typically get cleaned up together */ private String connectionID; - private int processorBatchSize = BufferConfig.DEFAULT_PROCESSOR_BATCH_SIZE; + private int processorBatchSize = BufferManager.DEFAULT_PROCESSOR_BATCH_SIZE; - private int connectorBatchSize = BufferConfig.DEFAULT_CONNECTOR_BATCH_SIZE; + private int connectorBatchSize = BufferManager.DEFAULT_CONNECTOR_BATCH_SIZE; private String userName; @@ -106,7 +106,6 @@ /** * Construct a new context. - * @param collectNodeStatistics TODO */ public CommandContext(Object processorID, String connectionID, String userName, Serializable commandPayload, String vdbName, String vdbVersion, Properties envProperties, boolean processDebug, boolean collectNodeStatistics) { Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/impl/ExecutionContextImpl.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/impl/ExecutionContextImpl.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/impl/ExecutionContextImpl.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -32,7 +32,7 @@ import org.teiid.dqp.internal.cache.DQPContextCache; import com.metamatrix.cache.Cache; -import com.metamatrix.common.buffer.impl.BufferConfig; +import com.metamatrix.common.buffer.BufferManager; import com.metamatrix.core.util.HashCodeUtil; /** @@ -68,7 +68,7 @@ private DQPContextCache contextCache; - private int batchSize = BufferConfig.DEFAULT_CONNECTOR_BATCH_SIZE; + private int batchSize = BufferManager.DEFAULT_CONNECTOR_BATCH_SIZE; private List<Exception> warnings = new LinkedList<Exception>(); public ExecutionContextImpl(String vdbName, String vdbVersion, String userName, Deleted: trunk/engine/src/main/java/org/teiid/dqp/internal/process/CacheResults.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/internal/process/CacheResults.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/CacheResults.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -1,68 +0,0 @@ -/* - * 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.Serializable; -import java.util.List; - -import com.metamatrix.common.buffer.TupleSourceID; -import com.metamatrix.query.analysis.AnalysisRecord; -import com.metamatrix.query.sql.lang.Command; -import com.metamatrix.query.sql.symbol.SingleElementSymbol; - -public class CacheResults implements Serializable { - private TupleSourceID results; - private AnalysisRecord analysisRecord; - private Command command; - - public CacheResults(TupleSourceID results){ - this.results = results; - } - - public TupleSourceID getResults() { - return results; - } - - public List<SingleElementSymbol> getElements() { - if (command == null) { - return null; - } - return command.getProjectedSymbols(); - } - - public Command getCommand() { - return command; - } - - public void setCommand(Command command) { - this.command = command; - } - - public AnalysisRecord getAnalysisRecord() { - return analysisRecord; - } - - public void setAnalysisRecord(AnalysisRecord analysisRecord) { - this.analysisRecord = analysisRecord; - } -} Modified: trunk/engine/src/test/java/com/metamatrix/common/buffer/BufferManagerFactory.java =================================================================== --- trunk/engine/src/test/java/com/metamatrix/common/buffer/BufferManagerFactory.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/test/java/com/metamatrix/common/buffer/BufferManagerFactory.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -52,10 +52,6 @@ if (INSTANCE == null) { BufferManagerImpl bufferMgr = new BufferManagerImpl(); Properties props = new Properties(); - props.setProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE, String.valueOf(Long.MAX_VALUE)); - props.setProperty(BufferManagerPropertyNames.SESSION_USE_PERCENTAGE, "100"); //$NON-NLS-1$ - props.setProperty(BufferManagerPropertyNames.LOG_STATS_INTERVAL, "0"); //$NON-NLS-1$ - props.setProperty(BufferManagerPropertyNames.MANAGEMENT_INTERVAL, "0"); //$NON-NLS-1$ bufferMgr.initialize(props); // Add unmanaged memory storage manager Modified: trunk/engine/src/test/java/com/metamatrix/common/buffer/storage/file/TestFileStorageManager.java =================================================================== --- trunk/engine/src/test/java/com/metamatrix/common/buffer/storage/file/TestFileStorageManager.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/test/java/com/metamatrix/common/buffer/storage/file/TestFileStorageManager.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -36,7 +36,7 @@ import junit.framework.TestCase; import com.metamatrix.api.exception.MetaMatrixComponentException; -import com.metamatrix.common.buffer.BufferManagerPropertyNames; +import com.metamatrix.common.buffer.BufferManager; import com.metamatrix.common.buffer.StorageManager; import com.metamatrix.common.buffer.TupleBatch; import com.metamatrix.common.buffer.TupleSourceID; @@ -61,23 +61,15 @@ } // Set up the fixture for this testcase: the tables for this test. - public StorageManager getStorageManager(String MAX_FILE_SIZE) { - try { - Properties resourceProps = new Properties(); - resourceProps.put(BufferManagerPropertyNames.BUFFER_STORAGE_DIRECTORY, UnitTestUtil.getTestScratchPath()); - if (MAX_FILE_SIZE != null) { - resourceProps.put(BufferManagerPropertyNames.MAX_FILE_SIZE, MAX_FILE_SIZE); - } - StorageManager sm = new FileStorageManager(); - sm.initialize(resourceProps); - return sm; - } catch(Exception e) { - e.printStackTrace(); - fail("Failure during storage manager initialization: " + e.getMessage()); //$NON-NLS-1$ - - // won't be called - return null; + public StorageManager getStorageManager(String MAX_FILE_SIZE) throws MetaMatrixComponentException { + Properties resourceProps = new Properties(); + resourceProps.put(BufferManager.BUFFER_STORAGE_DIRECTORY, UnitTestUtil.getTestScratchPath()); + if (MAX_FILE_SIZE != null) { + resourceProps.put(BufferManager.MAX_FILE_SIZE, MAX_FILE_SIZE); } + StorageManager sm = new FileStorageManager(); + sm.initialize(resourceProps); + return sm; } public static TupleBatch exampleBatch(int begin, int end) { @@ -146,17 +138,13 @@ } } - public void helpTestMultiThreaded(int OPEN_FILES, int NUM_THREADS, int NUM_BATCHES, int BATCH_SIZE) { + public void helpTestMultiThreaded(int OPEN_FILES, int NUM_THREADS, int NUM_BATCHES, int BATCH_SIZE) throws MetaMatrixComponentException { Properties resourceProps = new Properties(); String nonExistentDirectory = UnitTestUtil.getTestScratchPath() + File.separator + "testMultiThread"; //$NON-NLS-1$ - resourceProps.put(BufferManagerPropertyNames.BUFFER_STORAGE_DIRECTORY, nonExistentDirectory); - resourceProps.put(BufferManagerPropertyNames.MAX_OPEN_FILES, "" + OPEN_FILES); //$NON-NLS-1$ + resourceProps.put(BufferManager.BUFFER_STORAGE_DIRECTORY, nonExistentDirectory); + resourceProps.put(BufferManager.MAX_OPEN_FILES, "" + OPEN_FILES); //$NON-NLS-1$ final StorageManager sm = getStorageManager(null); - try { - sm.initialize(resourceProps); - } catch(MetaMatrixComponentException e) { - fail("Unexpected exception during initialization: " + e.getMessage()); //$NON-NLS-1$ - } + sm.initialize(resourceProps); // Create threads AddGetWorker[] threads = new AddGetWorker[NUM_THREADS]; @@ -198,17 +186,13 @@ } - public void helpTestRandomThreads(int OPEN_FILES, int NUM_THREADS, int NUM_BATCHES, int BATCH_SIZE, int RANDOM_OPS) { + public void helpTestRandomThreads(int OPEN_FILES, int NUM_THREADS, int NUM_BATCHES, int BATCH_SIZE, int RANDOM_OPS) throws MetaMatrixComponentException { Properties resourceProps = new Properties(); String nonExistentDirectory = UnitTestUtil.getTestScratchPath() + File.separator + "testMultiThread"; //$NON-NLS-1$ - resourceProps.put(BufferManagerPropertyNames.BUFFER_STORAGE_DIRECTORY, nonExistentDirectory); - resourceProps.put(BufferManagerPropertyNames.MAX_OPEN_FILES, "" + OPEN_FILES); //$NON-NLS-1$ + resourceProps.put(BufferManager.BUFFER_STORAGE_DIRECTORY, nonExistentDirectory); + resourceProps.put(BufferManager.MAX_OPEN_FILES, "" + OPEN_FILES); //$NON-NLS-1$ final StorageManager sm = getStorageManager(null); - try { - sm.initialize(resourceProps); - } catch(MetaMatrixComponentException e) { - fail("Unexpected exception during initialization: " + e.getMessage()); //$NON-NLS-1$ - } + sm.initialize(resourceProps); // Create threads RandomAccessWorker[] threads = new RandomAccessWorker[NUM_THREADS]; @@ -296,7 +280,7 @@ Properties resourceProps = new Properties(); String nonExistentDirectory = UnitTestUtil.getTestScratchPath() + File.separator + "GONZO"; //$NON-NLS-1$ - resourceProps.put(BufferManagerPropertyNames.BUFFER_STORAGE_DIRECTORY, nonExistentDirectory); + resourceProps.put(BufferManager.BUFFER_STORAGE_DIRECTORY, nonExistentDirectory); StorageManager sm = new FileStorageManager(); sm.initialize(resourceProps); @@ -327,27 +311,27 @@ } // Test with more open files than number of threads - public void testMultiThreaded1() { + public void testMultiThreaded1() throws Exception { helpTestMultiThreaded(6, 5, 5, 100); } // Test with fewer open files than number of threads - public void testMultiThreaded2() { + public void testMultiThreaded2() throws Exception { helpTestMultiThreaded(1, 2, 5, 100); } // Test 1 random thread - public void testRandomThreads1() { + public void testRandomThreads1() throws Exception { helpTestRandomThreads(10, 1, 5, 20, 30); } // Test several random threads - public void testRandomThreads2() { + public void testRandomThreads2() throws Exception { helpTestRandomThreads(10, 5, 5, 20, 30); } // Test several random threads, low open files - public void testRandomThreads3() { + public void testRandomThreads3() throws Exception { helpTestRandomThreads(2, 4, 5, 20, 30); } Modified: trunk/engine/src/test/java/com/metamatrix/query/processor/TestProcessor.java =================================================================== --- trunk/engine/src/test/java/com/metamatrix/query/processor/TestProcessor.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/test/java/com/metamatrix/query/processor/TestProcessor.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -226,8 +226,8 @@ public static void doProcess(ProcessorPlan plan, ProcessorDataManager dataManager, List[] expectedResults, CommandContext context) throws Exception { BufferManagerImpl bufferMgr = (BufferManagerImpl)BufferManagerFactory.getStandaloneBufferManager(); - bufferMgr.getConfig().setProcessorBatchSize(context.getProcessorBatchSize()); - bufferMgr.getConfig().setConnectorBatchSize(context.getProcessorBatchSize()); + bufferMgr.setProcessorBatchSize(context.getProcessorBatchSize()); + bufferMgr.setConnectorBatchSize(context.getProcessorBatchSize()); context.getNextRand(0); TupleBuffer id = null; try { Modified: trunk/engine/src/test/java/com/metamatrix/query/processor/TestVirtualDepJoin.java =================================================================== --- trunk/engine/src/test/java/com/metamatrix/query/processor/TestVirtualDepJoin.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/test/java/com/metamatrix/query/processor/TestVirtualDepJoin.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -38,7 +38,6 @@ import com.metamatrix.api.exception.MetaMatrixComponentException; import com.metamatrix.api.exception.MetaMatrixException; import com.metamatrix.common.buffer.BufferManager; -import com.metamatrix.common.buffer.BufferManagerPropertyNames; import com.metamatrix.common.buffer.TupleBuffer; import com.metamatrix.common.buffer.TupleSourceNotFoundException; import com.metamatrix.common.buffer.impl.BufferManagerImpl; @@ -495,12 +494,8 @@ private BufferManager createCustomBufferMgr(int batchSize) throws MetaMatrixComponentException { BufferManagerImpl bufferMgr = new BufferManagerImpl(); Properties props = new Properties(); - props.setProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE, String.valueOf(Long.MAX_VALUE)); - props.setProperty(BufferManagerPropertyNames.SESSION_USE_PERCENTAGE, "100"); //$NON-NLS-1$ - props.setProperty(BufferManagerPropertyNames.LOG_STATS_INTERVAL, "0"); //$NON-NLS-1$ - props.setProperty(BufferManagerPropertyNames.MANAGEMENT_INTERVAL, "0"); //$NON-NLS-1$ - props.setProperty(BufferManagerPropertyNames.PROCESSOR_BATCH_SIZE, String.valueOf(batchSize)); - props.setProperty(BufferManagerPropertyNames.CONNECTOR_BATCH_SIZE, String.valueOf(batchSize)); + props.setProperty(BufferManager.PROCESSOR_BATCH_SIZE, String.valueOf(batchSize)); + props.setProperty(BufferManager.CONNECTOR_BATCH_SIZE, String.valueOf(batchSize)); bufferMgr.initialize(props); // Add unmanaged memory storage manager Modified: trunk/engine/src/test/java/com/metamatrix/query/processor/relational/NodeTestUtil.java =================================================================== --- trunk/engine/src/test/java/com/metamatrix/query/processor/relational/NodeTestUtil.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/test/java/com/metamatrix/query/processor/relational/NodeTestUtil.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -26,7 +26,6 @@ import com.metamatrix.api.exception.MetaMatrixComponentException; import com.metamatrix.common.buffer.BufferManager; -import com.metamatrix.common.buffer.BufferManagerPropertyNames; import com.metamatrix.common.buffer.StorageManager; import com.metamatrix.common.buffer.impl.BufferManagerImpl; import com.metamatrix.common.buffer.impl.MemoryStorageManager; @@ -41,9 +40,8 @@ // Get the properties for BufferManager Properties bmProps = new Properties(); - bmProps.setProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE, "" + bytesAvailable); //$NON-NLS-1$ - bmProps.setProperty(BufferManagerPropertyNames.PROCESSOR_BATCH_SIZE, "" + procBatchSize); //$NON-NLS-1$ - bmProps.setProperty(BufferManagerPropertyNames.CONNECTOR_BATCH_SIZE, "" + connectorBatchSize); //$NON-NLS-1$ + bmProps.setProperty(BufferManager.PROCESSOR_BATCH_SIZE, "" + procBatchSize); //$NON-NLS-1$ + bmProps.setProperty(BufferManager.CONNECTOR_BATCH_SIZE, "" + connectorBatchSize); //$NON-NLS-1$ return createBufferManager(bmProps); } @@ -51,14 +49,12 @@ // Get the properties for BufferManager Properties bmProps = new Properties(); - bmProps.setProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE, "" + bytesAvailable); //$NON-NLS-1$ - bmProps.setProperty(BufferManagerPropertyNames.PROCESSOR_BATCH_SIZE, "" + procBatchSize); //$NON-NLS-1$ + bmProps.setProperty(BufferManager.PROCESSOR_BATCH_SIZE, "" + procBatchSize); //$NON-NLS-1$ return createBufferManager(bmProps); } static BufferManager createBufferManager(Properties bmProps) { BufferManagerImpl bufferManager = new BufferManagerImpl(); - bmProps.setProperty(BufferManagerPropertyNames.MANAGEMENT_INTERVAL, "0"); //$NON-NLS-1$ try { bufferManager.initialize(bmProps); } catch (MetaMatrixComponentException e) { Modified: trunk/engine/src/test/java/com/metamatrix/query/processor/relational/TestGroupingNode.java =================================================================== --- trunk/engine/src/test/java/com/metamatrix/query/processor/relational/TestGroupingNode.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/test/java/com/metamatrix/query/processor/relational/TestGroupingNode.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -207,7 +207,7 @@ // Same as test2, but uses processor batch size smaller than number of groups public void test3() throws Exception { BufferManager mgr = BufferManagerFactory.getStandaloneBufferManager(); - ((BufferManagerImpl)mgr).getConfig().setProcessorBatchSize(5); + ((BufferManagerImpl)mgr).setProcessorBatchSize(5); GroupingNode node = getExampleGroupingNode(); CommandContext context = new CommandContext("pid", "test", null, null, null); //$NON-NLS-1$ //$NON-NLS-2$ @@ -308,7 +308,7 @@ private void helpTestLookupFunctionInAggregate(int batchSize) throws Exception { BufferManager mgr = BufferManagerFactory.getStandaloneBufferManager(); - ((BufferManagerImpl)mgr).getConfig().setProcessorBatchSize(batchSize); + ((BufferManagerImpl)mgr).setProcessorBatchSize(batchSize); // Set up GroupingNode node = new GroupingNode(1); Modified: trunk/engine/src/test/java/com/metamatrix/query/processor/relational/TestJoinNode.java =================================================================== --- trunk/engine/src/test/java/com/metamatrix/query/processor/relational/TestJoinNode.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/engine/src/test/java/com/metamatrix/query/processor/relational/TestJoinNode.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -214,7 +214,10 @@ public void helpTestJoin() throws MetaMatrixComponentException, MetaMatrixProcessingException { for (int batchSize : new int[] {1, 10, leftTuples.length, 100}) { - helpCreateJoin(); + helpCreateJoin(); + if (batchSize == 0) { + continue; + } helpTestJoinDirect(expected, batchSize); List[] temp = leftTuples; leftTuples = rightTuples; Modified: trunk/runtime/src/main/java/com/metamatrix/dqp/embedded/admin/DQPRuntimeStateAdminImpl.java =================================================================== --- trunk/runtime/src/main/java/com/metamatrix/dqp/embedded/admin/DQPRuntimeStateAdminImpl.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/runtime/src/main/java/com/metamatrix/dqp/embedded/admin/DQPRuntimeStateAdminImpl.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -34,7 +34,6 @@ import org.teiid.adminapi.AdminProcessingException; import org.teiid.adminapi.Cache; import org.teiid.adminapi.ConnectorBinding; -import org.teiid.adminapi.EmbeddedLogger; import org.teiid.adminapi.Request; import org.teiid.adminapi.RuntimeStateAdmin; import org.teiid.dqp.internal.process.DQPWorkContext; @@ -42,7 +41,6 @@ import com.metamatrix.admin.objects.MMRequest; import com.metamatrix.api.exception.MetaMatrixComponentException; import com.metamatrix.common.application.exception.ApplicationLifecycleException; -import com.metamatrix.common.log.LogManager; import com.metamatrix.dqp.embedded.DQPEmbeddedPlugin; import com.metamatrix.dqp.message.AtomicRequestID; import com.metamatrix.dqp.message.RequestID; Modified: trunk/runtime/src/main/java/com/metamatrix/dqp/embedded/services/EmbeddedBufferService.java =================================================================== --- trunk/runtime/src/main/java/com/metamatrix/dqp/embedded/services/EmbeddedBufferService.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/runtime/src/main/java/com/metamatrix/dqp/embedded/services/EmbeddedBufferService.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -34,7 +34,6 @@ import com.metamatrix.common.application.exception.ApplicationInitializationException; import com.metamatrix.common.application.exception.ApplicationLifecycleException; import com.metamatrix.common.buffer.BufferManager; -import com.metamatrix.common.buffer.BufferManagerPropertyNames; import com.metamatrix.common.buffer.StorageManager; import com.metamatrix.common.buffer.impl.BufferManagerImpl; import com.metamatrix.common.buffer.impl.FileStorageManager; @@ -51,12 +50,6 @@ */ public class EmbeddedBufferService extends EmbeddedBaseDQPService implements BufferService { - // Constants - private static final String DEFAULT_MANAGEMENT_INTERVAL = "500"; //$NON-NLS-1$ - private static final String DEFAULT_LOG_STATS_INTERVAL = "60000"; //$NON-NLS-1$ // every minute - private static final String DEFAULT_SESSION_USE_PERCENTAGE = "100"; //$NON-NLS-1$ - private static final String DEFAULT_MAX_OPEN_FILES = "10"; //$NON-NLS-1$ - // Instance private BufferManagerImpl bufferMgr; private File bufferDir; @@ -95,22 +88,14 @@ boolean useDisk = configurationSvc.useDiskBuffering(); bufferDir = configurationSvc.getDiskBufferDirectory(); - String memAvail = configurationSvc.getBufferMemorySize(); String processorBatchSize = configurationSvc.getProcessorBatchSize(); String connectorBatchSize = configurationSvc.getConnectorBatchSize(); // Set up buffer configuration properties Properties bufferProps = new Properties(); - bufferProps.setProperty(BufferManagerPropertyNames.SESSION_USE_PERCENTAGE, DEFAULT_SESSION_USE_PERCENTAGE); - bufferProps.setProperty(BufferManagerPropertyNames.LOG_STATS_INTERVAL, DEFAULT_LOG_STATS_INTERVAL); - bufferProps.setProperty(BufferManagerPropertyNames.MANAGEMENT_INTERVAL, DEFAULT_MANAGEMENT_INTERVAL); - bufferProps.setProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE, memAvail); - bufferProps.setProperty(BufferManagerPropertyNames.BUFFER_STORAGE_DIRECTORY, bufferDir.getCanonicalPath()); - bufferProps.setProperty(BufferManagerPropertyNames.PROCESSOR_BATCH_SIZE, processorBatchSize); - bufferProps.setProperty(BufferManagerPropertyNames.CONNECTOR_BATCH_SIZE, connectorBatchSize); - // These are not set yet.. - //CONNECTOR_BATCH_SIZE - //PROCESSOR_BATCH_SIZE + bufferProps.setProperty(BufferManager.BUFFER_STORAGE_DIRECTORY, bufferDir.getCanonicalPath()); + bufferProps.setProperty(BufferManager.PROCESSOR_BATCH_SIZE, processorBatchSize); + bufferProps.setProperty(BufferManager.CONNECTOR_BATCH_SIZE, connectorBatchSize); // Construct and initialize the buffer manager this.bufferMgr = new BufferManagerImpl(); @@ -120,8 +105,7 @@ if(useDisk) { // Get the properties for FileStorageManager and create. Properties fsmProps = new Properties(); - fsmProps.setProperty(BufferManagerPropertyNames.BUFFER_STORAGE_DIRECTORY, bufferDir.getCanonicalPath()); - fsmProps.setProperty(BufferManagerPropertyNames.MAX_OPEN_FILES, DEFAULT_MAX_OPEN_FILES); + fsmProps.setProperty(BufferManager.BUFFER_STORAGE_DIRECTORY, bufferDir.getCanonicalPath()); StorageManager fsm = new FileStorageManager(); fsm.initialize(fsmProps); this.bufferMgr.setStorageManager(fsm); Modified: trunk/runtime/src/test/java/com/metamatrix/dqp/service/buffer/TestLocalBufferService.java =================================================================== --- trunk/runtime/src/test/java/com/metamatrix/dqp/service/buffer/TestLocalBufferService.java 2009-12-24 05:03:40 UTC (rev 1704) +++ trunk/runtime/src/test/java/com/metamatrix/dqp/service/buffer/TestLocalBufferService.java 2009-12-24 05:21:00 UTC (rev 1705) @@ -28,8 +28,8 @@ import com.metamatrix.common.application.ApplicationEnvironment; import com.metamatrix.common.application.exception.ApplicationInitializationException; -import com.metamatrix.common.buffer.impl.BufferConfig; import com.metamatrix.common.buffer.impl.BufferManagerImpl; +import com.metamatrix.common.buffer.impl.FileStorageManager; import com.metamatrix.core.util.UnitTestUtil; import com.metamatrix.dqp.embedded.DQPEmbeddedProperties; import com.metamatrix.dqp.embedded.EmbeddedTestUtil; @@ -90,9 +90,7 @@ assertTrue(cs.useDiskBuffering()); BufferManagerImpl mgr = (BufferManagerImpl) svc.getBufferManager(); - BufferConfig config = mgr.getConfig(); - assertEquals("Did not get expected memory level", 96000000L, config.getTotalAvailableMemory()); //$NON-NLS-1$ - assertTrue(config.getBufferStorageDirectory().endsWith(cs.getDiskBufferDirectory().getName())); + assertTrue(((FileStorageManager)mgr.getStorageManager()).getDirectory().endsWith(cs.getDiskBufferDirectory().getName())); } public void testCheckMemPropertyGotSet2() throws Exception { @@ -110,10 +108,6 @@ // all the properties are set assertFalse(cs.useDiskBuffering()); - - BufferManagerImpl mgr = (BufferManagerImpl) svc.getBufferManager(); - BufferConfig config = mgr.getConfig(); - assertEquals("Did not get expected memory level", 64000000L, config.getTotalAvailableMemory()); //$NON-NLS-1$ } }