mergesort.java

derby database source code.good for you.

	/**
	Check the column ordering against the template, making
	sure that each column is present in the template,
	implements Orderable, and is not mentioned more than
	once.  Intended to be called as part of a sanity check.
	**/
	protected boolean checkColumnOrdering(
    DataValueDescriptor[]   template, 
    ColumnOrdering          columnOrdering[])
	{
		// Allocate an array to check that each column mentioned only once.
		int templateNColumns = template.length;
		boolean seen[] = new boolean[templateNColumns];

		// Check each column ordering.
		for (int i = 0; i < columnOrdering.length; i++)
		{
			int colid = columnOrdering[i].getColumnId();

			// Check that the column id is valid.
			if (colid < 0 || colid >= templateNColumns)
				return false;
			
			// Check that the column isn't mentioned more than once.
			if (seen[colid])
				return false;
			seen[colid] = true;

			Object columnVal = 
                RowUtil.getColumn(template, (FormatableBitSet) null, colid);

			if (!(columnVal instanceof Orderable))
				return false;
		}

		return true;
	}

	/**
	Check that the columns in the row agree with the columns
	in the template, both in number and in type.
	<p>
	XXX (nat) Currently checks that the classes implementing
	each column are the same -- is this right?
	**/
	void checkColumnTypes(DataValueDescriptor[] row)
		throws StandardException
	{
		int nCols = row.length;
		if (template.length != nCols)
		{
            if (SanityManager.DEBUG)
            {
                SanityManager.THROWASSERT(
                    "template.length (" + template.length +
                    ") expected to be = to nCols (" +
                    nCols + ")");
            }
            throw StandardException.newException(
                    SQLState.SORT_TYPE_MISMATCH);
		}

        if (SanityManager.DEBUG)
        {
            for (int colid = 0; colid < nCols; colid++)
            {
                Object col1 = row[colid];
                Object col2 = template[colid];
                if (col1 == null)
				{
					SanityManager.THROWASSERT(
						"col[" + colid + "]  is null");
				}
						
                if (!(col1 instanceof CloneableObject))
				{
					SanityManager.THROWASSERT(
						"col[" + colid + "] (" +col1.getClass().getName()+
						") is not a CloneableObject.");
				}

                if (col1.getClass() != col2.getClass())
                {
                    SanityManager.THROWASSERT(
                        "col1.getClass() (" + col1.getClass() +
                        ") expected to be the same as col2.getClass() (" +
                        col2.getClass() + ")");
                }
            }
        }
	}

	int compare(
    DataValueDescriptor[] r1, 
    DataValueDescriptor[] r2)
		throws StandardException
	{
		// Get the number of columns we have to compare.
		int colsToCompare = this.columnOrdering.length;
        int r;

		// Compare the columns specified in the column
		// ordering array.
        for (int i = 0; i < colsToCompare; i++)
        {
			// Get columns to compare.
            int colid = this.columnOrderingMap[i];

			// If the columns don't compare equal, we're done.
			// Return the sense of the comparison.
			if ((r = r1[colid].compare(r2[colid])) 
                    != 0)
			{
				if (this.columnOrderingAscendingMap[i])
					return r;
				else
					return -r;
			}
		}

		// We made it through all the columns, and they must have
		// all compared equal.  So return that the rows compare equal.
		return 0;
	}

	/**
	Go from the CLOSED to the INITIALIZED state.
	**/
	public void initialize(
    DataValueDescriptor[]   template,
    ColumnOrdering          columnOrdering[],
    SortObserver            sortObserver,
    boolean                 alreadyInOrder,
    long                    estimatedRows,
    int                     sortBufferMax)
        throws StandardException
	{
        if (SanityManager.DEBUG)
        {
    		SanityManager.ASSERT(state == STATE_CLOSED);
    	}

		// Make sure the column ordering makes sense
        if (SanityManager.DEBUG)
        {
    		SanityManager.ASSERT(checkColumnOrdering(template, columnOrdering),
	    		"column ordering error");
	    }

		// Set user-defined parameters.
		this.template = template;
		this.columnOrdering = columnOrdering;
		this.sortObserver = sortObserver;
		this.alreadyInOrder = alreadyInOrder;

        // Cache results of columnOrdering calls, results are not allowed
        // to change throughout a sort.
        columnOrderingMap          = new int[columnOrdering.length];
        columnOrderingAscendingMap = new boolean[columnOrdering.length];
        for (int i = 0; i < columnOrdering.length; i++)
        {
            columnOrderingMap[i] = columnOrdering[i].getColumnId();
            columnOrderingAscendingMap[i] = columnOrdering[i].getIsAscending();
        }

		// No inserter or scan yet.
		this.inserter = null;
		this.scan = null;

		// We don't have any merge runs.
		this.mergeRuns = null;
		this.sortBuffer = null;
		this.sortBufferMax = sortBufferMax;

        if (estimatedRows > sortBufferMax)
			sortBufferMin = sortBufferMax;
		else
			sortBufferMin = (int)estimatedRows;
		if (SanityManager.DEBUG)
        {
            if (SanityManager.DEBUG_ON("testSort"))
                sortBufferMin = sortBufferMax;
        }

		this.state = STATE_INITIALIZED;
	}

	/**
	An inserter is closing.
	**/
	void doneInserting(MergeInserter inserter,
		SortBuffer sortBuffer, Vector mergeRuns)
	{
        if (SanityManager.DEBUG)
        {
    		SanityManager.ASSERT(state == STATE_INSERTING);
    	}

		this.sortBuffer = sortBuffer;
		this.mergeRuns = mergeRuns;
		this.inserter = null;

		this.state = STATE_DONE_INSERTING;
	}

	void doneScanning(Scan scan, SortBuffer sortBuffer)
	{
		if (SanityManager.DEBUG)
		{
			// Make sure the scan we're getting back is the one we gave out

			if (this.scan != scan)
    			SanityManager.THROWASSERT("this.scan = " + this.scan 
										  + " scan = " + scan);
		}

		this.sortBuffer = sortBuffer;
		this.scan = null;

		this.state = STATE_DONE_SCANNING;
	}

	void doneScanning(Scan scan, SortBuffer sortBuffer,
		Vector mergeRuns)
	{
		this.mergeRuns = mergeRuns;

		doneScanning(scan, sortBuffer);
	}


	/**
	Get rid of the merge runs, if there are any.
	Must not cause any errors because it's called
	during error processing.
	**/
	void dropMergeRuns(TransactionManager tran)
	{
		if (mergeRuns != null)
		{
			Enumeration e = mergeRuns.elements();

			try 
			{
				Transaction rawTran = tran.getRawStoreXact();
				long segmentId = StreamContainerHandle.TEMPORARY_SEGMENT;

				while (e.hasMoreElements())
				{
					long containerId = ((Long) e.nextElement()).longValue();
					rawTran.dropStreamContainer(segmentId, containerId);
				}
			}
			catch (StandardException se)
			{
				// Ignore problems with dropping, worst case
				// the raw store will clean up at reboot.
			}
			mergeRuns = null;
		}
	}

	/* DEBUG (nat)
	void printRunInfo(TransactionController tran)
		throws StandardException
	{
		java.util.Enumeration e = mergeRuns.elements();
		while (e.hasMoreElements())
		{
			long conglomid = ((Long) e.nextElement()).longValue();
			ScanController sc = tran.openScan(conglomid, false,
									false, null, null, 0, null,
									null, 0);
			System.out.println("Merge run: conglomid=" + conglomid);
			while (sc.next())
			{
				sc.fetch(template);
				System.out.println(template);
			}
			sc.close();
		}
	}
	*/

	private void multiStageMerge(TransactionManager tran)
		throws StandardException
	{
		Enumeration e;
		//int iterations = 0; // DEBUG (nat)
		int maxMergeRuns = sortBuffer.capacity();

		if (maxMergeRuns > ExternalSortFactory.DEFAULT_MAX_MERGE_RUN)
			maxMergeRuns = ExternalSortFactory.DEFAULT_MAX_MERGE_RUN;

		Vector subset;
		Vector leftovers;

		while (mergeRuns.size() > maxMergeRuns)
		{
			// Move maxMergeRuns elements from the merge runs
			// vector into a subset, leaving the rest.
			subset = new Vector(maxMergeRuns);
			leftovers = new Vector(mergeRuns.size() - maxMergeRuns);
			e = mergeRuns.elements();
			while (e.hasMoreElements())
			{
				Long containerId = (Long) e.nextElement();
				if (subset.size() < maxMergeRuns)
					subset.addElement(containerId);
				else
					leftovers.addElement(containerId);
			}

			/* DEBUG (nat)
			iterations++;
				System.out.println(subset.size() + " elements in subset");
				System.out.println(leftovers.size() + " elements in leftovers");
				System.out.println(mergeRuns.size() + " elements in mergeRuns");
				System.out.println("maxMergeRuns is " + maxMergeRuns);
				System.out.println("iterations = " + iterations);
			if (subset.size() == 0)
			{
				System.exit(1);
			}
			*/

			mergeRuns = leftovers;

			// Open a merge scan on the subset.
			MergeScanRowSource msRowSource = 
				new MergeScanRowSource(this, tran, sortBuffer, subset, sortObserver, false);

			if (!msRowSource.init(tran))
            {
                throw StandardException.newException(
                        SQLState.SORT_COULD_NOT_INIT);
            }

			// Create and open another temporary stream conglomerate
			// which will become
			// a merge run made up with the merged runs from the subset.
			Transaction rawTran = tran.getRawStoreXact();
			int segmentId = StreamContainerHandle.TEMPORARY_SEGMENT;
			long id = rawTran.addAndLoadStreamContainer(segmentId,
				properties, msRowSource);

			mergeRuns.addElement(new Long(id));

			// Drop the conglomerates in the merge subset
			e = subset.elements();
			while (e.hasMoreElements())
			{
				Long containerId = (Long) e.nextElement();
				rawTran.dropStreamContainer(segmentId, containerId.longValue());
			}
		}
	}

	/**
	Remove all the rows from the sort buffer and store them
	in a temporary conglomerate.  The temporary conglomerate
	is a "merge run".  Returns the container id of the
	merge run.
	**/
	long createMergeRun(TransactionManager tran, SortBuffer sortBuffer)
		throws StandardException
	{
		// this sort buffer is not a scan and is not tracked by any
		// TransactionManager. 
		SortBufferRowSource rowSource =
			new SortBufferRowSource(sortBuffer, (TransactionManager)null, sortObserver, true, sortBufferMax); 

		// Create a temporary stream conglomerate...
		Transaction rawTran = tran.getRawStoreXact();  // get raw transaction
		int segmentId = StreamContainerHandle.TEMPORARY_SEGMENT;
		long id = rawTran.addAndLoadStreamContainer(segmentId,
			properties, rowSource);

		// Don't close the sortBuffer, we just emptied it, the caller may reuse
		// that sortBuffer for the next run.
		rowSource = null;

		return id;
	}
}