groupedaggregateresultset.java

derby database source code.good for you.

				return null;
			}

		    ExecIndexRow nextRow = getNextRowFromRS();

			/* Drain and merge rows until we find new distinct values for the grouping columns. */
			while (nextRow != null)
			{
				/* We found a new set of values for the grouping columns.  
				 * Update the current row and return this group. 
				 */
				if (! sameGroupingValues(currSortedRow, nextRow))
				{
					ExecIndexRow result = currSortedRow;

					/* Save a clone of the new row so that it doesn't get overwritten */
					currSortedRow = (ExecIndexRow) nextRow.getClone();
					initializeVectorAggregation(currSortedRow);

					nextTime += getElapsedMillis(beginTime);
					rowsReturned++;
					return finishAggregation(result);
				}
				else
				{
					/* Same group - initialize the new row and then merge the aggregates */
					initializeVectorAggregation(nextRow);
					mergeVectorAggregates(nextRow, currSortedRow);
				}

				// Get the next row
				nextRow = getNextRowFromRS();
			}

			// We've drained the source, so no more rows to return
			ExecIndexRow result = currSortedRow;
			currSortedRow = null;
			nextTime += getElapsedMillis(beginTime);
			return finishAggregation(result);
		}
		else
		{
		    ExecIndexRow sortResult = null;

	        if ((sortResult = getNextRowFromRS()) != null)
			{
				setCurrentRow(sortResult);
			}

			/*
			** Only finish the aggregation
			** if we have a return row.  We don't generate
			** a row on a vector aggregate unless there was
			** a group.
			*/
			if (sortResult != null)
			{
				sortResult = finishAggregation(sortResult);
				currentRow = sortResult;
			}

			if (sortResult != null)
			{
				rowsReturned++;
			}

			nextTime += getElapsedMillis(beginTime);
		    return sortResult;
		}
	}

	/**
	 * Return whether or not the new row has the same values for the 
	 * grouping columns as the current row.  (This allows us to process in-order
	 * group bys without a sorter.)
	 *
	 * @param currRow	The current row.
	 * @param newRow	The new row.
	 *
	 * @return	Whether or not to filter out the new row has the same values for the 
	 *			grouping columns as the current row.
	 *
	 * @exception StandardException thrown on failure to get row location
	 */
	private boolean sameGroupingValues(ExecRow currRow, ExecRow newRow)
		throws StandardException
	{
		for (int index = 0; index < order.length; index++)
		{
			DataValueDescriptor currOrderable = currRow.getColumn(order[index].getColumnId() + 1);
			DataValueDescriptor newOrderable = newRow.getColumn(order[index].getColumnId() + 1);
			if (! (currOrderable.compare(DataValueDescriptor.ORDER_OP_EQUALS, newOrderable, true, true)))
			{
				return false;
			}
		}
		return true;
	}

	/**
	 * If the result set has been opened,
	 * close the open scan.
	 *
	 * @exception StandardException thrown on error
	 */
	public void	close() throws StandardException
	{
		beginTime = getCurrentTimeMillis();
		if ( isOpen )
	    {
			// we don't want to keep around a pointer to the
			// row ... so it can be thrown away.
			// REVISIT: does this need to be in a finally
			// block, to ensure that it is executed?
		    clearCurrentRow();
			if (closeCleanup != null) {
				closeCleanup.invoke(activation); // let activation tidy up
			}

			currentRow = null;
			sortResultRow = null;
			sourceExecIndexRow = null;
			closeSource();

			if (dropDistinctAggSort)
			{
				tc.dropSort(distinctAggSortId);
				dropDistinctAggSort = false;
			}

			if (dropGenericSort)
			{
				tc.dropSort(genericSortId);
				dropGenericSort = false;
			}
			super.close();
		}
		else
			if (SanityManager.DEBUG)
				SanityManager.DEBUG("CloseRepeatInfo","Close of SortResultSet repeated");

		closeTime += getElapsedMillis(beginTime);

		isOpen = false;
	}

	/**
	 * Return the total amount of time spent in this ResultSet
	 *
	 * @param type	CURRENT_RESULTSET_ONLY - time spent only in this ResultSet
	 *				ENTIRE_RESULTSET_TREE  - time spent in this ResultSet and below.
	 *
	 * @return long		The total amount of time spent (in milliseconds).
	 */
	public long getTimeSpent(int type)
	{
		long totTime = constructorTime + openTime + nextTime + 
						closeTime;

		if (type == NoPutResultSet.CURRENT_RESULTSET_ONLY)
		{
			return	totTime - originalSource.getTimeSpent(ENTIRE_RESULTSET_TREE);
		}
		else
		{
			return totTime;
		}
	}

	///////////////////////////////////////////////////////////////////////////////
	//
	// CursorResultSet interface
	//
	///////////////////////////////////////////////////////////////////////////////

	/**
	 * This result set has its row location from
	 * the last fetch done. If the cursor is closed,
	 * a null is returned.
	 *
	 * @see CursorResultSet
	 *
	 * @return the row location of the current cursor row.
	 * @exception StandardException thrown on failure to get row location
	 */
	public RowLocation getRowLocation() throws StandardException
	{
		if (! isOpen) return null;

		// REVISIT: could we reuse the same rowlocation object
		// across several calls?
		RowLocation rl;
		rl = scanController.newRowLocationTemplate();
		scanController.fetchLocation(rl);
		return rl;
	}

	/**
	 * This result set has its row from the last fetch done. 
	 * If the cursor is closed, a null is returned.
	 *
	 * @see CursorResultSet
	 *
	 * @return the last row returned;
	 * @exception StandardException thrown on failure.
	 */
	/* RESOLVE - this should return activation.getCurrentRow(resultSetNumber),
	 * once there is such a method.  (currentRow is redundant)
	 */
	public ExecRow getCurrentRow() throws StandardException 
	{
		if (SanityManager.DEBUG)
			SanityManager.ASSERT(isOpen, "SortResultSet expected to be open");

		return currentRow;
	}

	///////////////////////////////////////////////////////////////////////////////
	//
	// SCAN ABSTRACTION UTILITIES
	//
	///////////////////////////////////////////////////////////////////////////////
	/**
	 * Get the next output row for processing
	 */
	private ExecIndexRow getNextRowFromRS()
		throws StandardException
	{
		return (scanController == null) ?
			getRowFromResultSet() :
			getRowFromSorter();
	}

	/**
	 * Get a row from the input result set.  
	 */	
	private ExecIndexRow getRowFromResultSet()
		throws StandardException
	{
		ExecRow					sourceRow;
		ExecIndexRow			inputRow = null;	

		if ((sourceRow = source.getNextRowCore()) != null)
		{
			rowsInput++;
			sourceExecIndexRow.execRowToExecIndexRow(sourceRow);
			inputRow = sourceExecIndexRow;
		}

		return inputRow;
	}


	/**
	 * Get a row from the sorter.  Side effects:
	 * sets currentRow.
	 */
	private ExecIndexRow getRowFromSorter()
		throws StandardException
	{
		ExecIndexRow			inputRow = null;	
		
		if (scanController.next())
		{
			// REMIND: HACKALERT we are assuming that result will
			// point to what sortResult is manipulating when
			// we complete the fetch.
			currentRow = sortResultRow;

			inputRow = getExecutionFactory().getIndexableRow(currentRow);

			scanController.fetch(inputRow.getRowArray());
		}
		return inputRow;
	}

	/**
	 * Close the source of whatever we have been scanning.
	 *
	 * @exception StandardException thrown on error
	 */
	public void	closeSource() throws StandardException
	{
		if (scanController == null)
		{
			/*
			** NOTE: do not null out source, we
			** may be opened again, in which case
			** we will open source again.
			*/
			source.close();
		}
		else
		{
			scanController.close();
			scanController = null;
		}
	}

	///////////////////////////////////////////////////////////////////////////////
	//
	// AGGREGATION UTILITIES
	//
	///////////////////////////////////////////////////////////////////////////////
	/**
	 * Run the aggregator initialization method for
	 * each aggregator in the row.  Accumulate the
	 * input column.  WARNING: initializiation performs
	 * accumulation -- no need to accumulate a row
	 * that has been passed to initialization.
	 *
	 * @param	row	the row to initialize
	 *
	 * @exception	standard cloudscape exception
	 */
	private void initializeVectorAggregation(ExecRow row)
		throws StandardException
	{
		int size = aggregates.length;

		if (SanityManager.DEBUG)
		{
			SanityManager.ASSERT(row != null, 
					"Null row passed to initializeVectorAggregation");
		}

		for (int i = 0; i < size; i++)
		{
			GenericAggregator currAggregate = aggregates[i];

			// initialize the aggregator
			currAggregate.initialize(row);

			// get the first value, accumulate it into itself
			currAggregate.accumulate(row, row);
		}
	}

	/**
	 * Run the aggregator merge method for
	 * each aggregator in the row.  
	 *
	 * @param	newRow	the row to merge
	 * @param	currRow the row to merge into
	 *
	 * @exception	standard cloudscape exception
	 */
	private void mergeVectorAggregates(ExecRow newRow, ExecRow currRow)
		throws StandardException
	{
		for (int i = 0; i < aggregates.length; i++)
		{
			GenericAggregator currAggregate = aggregates[i];

			// merge the aggregator
			currAggregate.merge(newRow, currRow);
		}
	}

}