Implemented #7331: Cost-based choice between nested loop join and hash join

Author: dyemanov

src/jrd/optimizer/InnerJoin.cpp

@@ -97,10 +97,10 @@ void InnerJoin::calculateStreamInfo()
 
 	for (auto innerStream : innerStreams)
 	{
-		streams.add(innerStream->stream);
-		csb->csb_rpt[innerStream->stream].activate();
+		streams.add(innerStream->number);
+		csb->csb_rpt[innerStream->number].activate();
 
-		Retrieval retrieval(tdbb, optimizer, innerStream->stream, false, false, sort, true);
+		Retrieval retrieval(tdbb, optimizer, innerStream->number, false, false, sort, true);
 		const auto candidate = retrieval.getInversion();
 
 		innerStream->baseCost = candidate->cost;
@@ -109,7 +109,7 @@ void InnerJoin::calculateStreamInfo()
 		innerStream->baseUnique = candidate->unique;
 		innerStream->baseNavigated = candidate->navigated;
 
-		csb->csb_rpt[innerStream->stream].deactivate();
+		csb->csb_rpt[innerStream->number].deactivate();
 	}
 
 	// Collect dependencies between every pair of streams
@@ -120,7 +120,7 @@ void InnerJoin::calculateStreamInfo()
 
 		for (const auto innerStream : innerStreams)
 		{
-			const StreamType testStream = innerStream->stream;
+			const StreamType testStream = innerStream->number;
 
 			if (baseStream != testStream)
 			{
@@ -145,7 +145,7 @@ void InnerJoin::calculateStreamInfo()
 	}
 
 	// Unless PLAN is enforced, sort the streams based on independency and cost
-	if (!plan && (innerStreams.getCount() > 1))
+	if (!plan && innerStreams.getCount() > 1)
 	{
 		StreamInfoList tempStreams;
 
@@ -172,23 +172,75 @@ void InnerJoin::calculateStreamInfo()
 
 void InnerJoin::estimateCost(unsigned position,
 							 const StreamInfo* stream,
-							 double* cost,
-							 double* resultingCardinality) const
+							 double& cost,
+							 double& cardinality)
 {
-	const auto sort = (position == 0 && sortPtr) ? *sortPtr : nullptr;
+	fb_assert(joinedStreams[position].number == stream->number);
+
+	const auto sort = (!position && sortPtr) ? *sortPtr : nullptr;
 
 	// Create the optimizer retrieval generation class and calculate
 	// which indexes will be used and the total estimated selectivity will be returned
-	Retrieval retrieval(tdbb, optimizer, stream->stream, false, false, sort, true);
+	Retrieval retrieval(tdbb, optimizer, stream->number, false, false, sort, true);
 	const auto candidate = retrieval.getInversion();
+	fb_assert(!position || candidate->dependencies);
+
+	// Calculate the relationship selectivity
+	double selectivity = candidate->selectivity;
+	if (selectivity < stream->baseSelectivity)
+		selectivity /= stream->baseSelectivity;
+
+	joinedStreams[position].selectivity = selectivity;
+
+	// Get the stream cardinality
+	const auto tail = &csb->csb_rpt[stream->number];
+	const auto streamCardinality = tail->csb_cardinality;
 
-	*cost = candidate->cost;
+	// Calculate the nested loop cost, it's our default option
+	const auto loopCost = candidate->cost * cardinality;
+	cost = loopCost;
 
-	// Calculate cardinality
-	const auto tail = &csb->csb_rpt[stream->stream];
-	const double cardinality = tail->csb_cardinality * candidate->selectivity;
+	if (position)
+	{
+		// Calculate the hashing cost. It's estimated as the hashed stream retrieval cost
+		// plus two cardinalities. Hashed stream cardinality means the cost of copying rows
+		// into the hash table and the outer cardinality represents probing the hash table.
+		const auto hashCardinality = stream->baseSelectivity * streamCardinality;
+		const auto hashCost = stream->baseCost + hashCardinality + cardinality;
+
+		if (hashCost <= loopCost && hashCardinality <= HashJoin::maxCapacity())
+		{
+			auto& equiMatches = joinedStreams[position].equiMatches;
+			fb_assert(!equiMatches.hasData());
 
-	*resultingCardinality = MAX(cardinality, MINIMUM_CARDINALITY);
+			// Scan the matches for possible equi-join conditions
+			for (const auto match : candidate->matches)
+			{
+				// Check whether we have an equivalence operation
+				if (!optimizer->checkEquiJoin(match))
+					continue;
+
+				// Check whether the match references priorly joined streams
+				const auto end = joinedStreams.begin() + position;
+				for (auto iter = joinedStreams.begin(); iter != end; ++iter)
+				{
+					if (match->containsStream(iter->number) &&
+						equiMatches.getCount() < equiMatches.getCapacity())
+					{
+						equiMatches.add(match);
+						break;
+					}
+				}
+			}
+
+			// Adjust the actual cost value, if hash joining is both possible and preferrable
+			if (equiMatches.hasData())
+				cost = hashCost;
+		}
+	}
+
+	const auto resultingCardinality = streamCardinality * candidate->selectivity;
+	cardinality = MAX(resultingCardinality, MINIMUM_CARDINALITY);
 }
 
 
@@ -231,7 +283,8 @@ bool InnerJoin::findJoinOrder()
 			{
 				if (!bestCount || innerStream->baseCost < bestCost)
 				{
-					joinedStreams[0].bestStream = innerStream->stream;
+					bestStreams.resize(1);
+					bestStreams.front().number = innerStream->number;
 					bestCount = 1;
 					bestCost = innerStream->baseCost;
 				}
@@ -270,11 +323,10 @@ bool InnerJoin::findJoinOrder()
 	}
 
 	// Mark streams as used
-	for (unsigned i = 0; i < bestCount; i++)
+	for (const auto& stream : bestStreams)
 	{
-		auto streamInfo = getStreamInfo(joinedStreams[i].bestStream);
+		auto streamInfo = getStreamInfo(stream.number);
 		streamInfo->used = true;
-		bestStreams.add(joinedStreams[i].bestStream);
 	}
 
 #ifdef OPT_DEBUG
@@ -299,11 +351,11 @@ void InnerJoin::findBestOrder(unsigned position,
 							  double cost,
 							  double cardinality)
 {
-	const auto tail = &csb->csb_rpt[stream->stream];
+	const auto tail = &csb->csb_rpt[stream->number];
 
 	// Do some initializations
 	tail->activate();
-	joinedStreams[position].number = stream->stream;
+	joinedStreams[position].reset(stream->number);
 
 	// Save the various flag bits from the optimizer block to reset its
 	// state after each test
@@ -312,13 +364,13 @@ void InnerJoin::findBestOrder(unsigned position,
 		streamFlags.add(innerStream->used);
 
 	// Compute delta and total estimate cost to fetch this stream
-	double positionCost = 0, positionCardinality = 0, newCost = 0, newCardinality = 0;
+	double positionCost = 0, positionCardinality = cardinality, newCost = 0, newCardinality = 0;
 
 	if (!plan)
 	{
-		estimateCost(position, stream, &positionCost, &positionCardinality);
-		newCost = cost + cardinality * positionCost;
-		newCardinality = positionCardinality * cardinality;
+		estimateCost(position, stream, positionCost, positionCardinality);
+		newCost = cost + positionCost;
+		newCardinality = cardinality * positionCardinality;
 	}
 
 	position++;
@@ -329,13 +381,7 @@ void InnerJoin::findBestOrder(unsigned position,
 	{
 		bestCount = position;
 		bestCost = newCost;
-
-		const auto end = joinedStreams.begin() + position;
-		for (auto iter = joinedStreams.begin(); iter != end; ++iter)
-		{
-			auto& joinedStream = *iter;
-			joinedStream.bestStream = joinedStream.number;
-		}
+		bestStreams.assign(joinedStreams.begin(), position);
 	}
 
 #ifdef OPT_DEBUG
@@ -398,7 +444,7 @@ void InnerJoin::findBestOrder(unsigned position,
 			IndexRelationship* processRelationship = processList.begin();
 			for (FB_SIZE_T index = 0; index < processList.getCount(); index++)
 			{
-				if (relationStreamInfo->stream == processRelationship[index].stream)
+				if (relationStreamInfo->number == processRelationship[index].stream)
 				{
 					// If the cost of this relationship is cheaper then remove the
 					// old relationship and add this one
@@ -445,25 +491,103 @@ void InnerJoin::findBestOrder(unsigned position,
 River* InnerJoin::formRiver()
 {
 	fb_assert(bestCount);
-	fb_assert(bestStreams.hasData());
+	fb_assert(bestStreams.getCount() == bestCount);
 
-	if (bestStreams.getCount() != innerStreams.getCount())
+	const auto orgSortPtr = sortPtr;
+
+	if (bestCount != innerStreams.getCount())
 		sortPtr = nullptr;
 
+	RecordSource* rsb;
+	StreamList streams;
 	HalfStaticArray<RecordSource*, OPT_STATIC_ITEMS> rsbs;
 
-	for (const auto stream : bestStreams)
+	for (const auto& stream : bestStreams)
 	{
-		const auto rsb = optimizer->generateRetrieval(stream, sortPtr, false, false);
+		// We use hash join instead of nested loop join if:
+		//  - stream has equivalence relationship(s) with the prior streams
+		//    (and hashing was estimated to be cheaper)
+		//  AND
+		//    - optimization for first rows is not requested
+		//    OR
+		//    - existing sort was not utilized using an index
+
+		if (rsbs.hasData() && // this is not the first stream
+			stream.equiMatches.hasData() &&
+			(!optimizer->favorFirstRows() || (orgSortPtr && *orgSortPtr)))
+		{
+			fb_assert(streams.hasData());
+
+			// Deactivate priorly joined streams
+			StreamStateHolder stateHolder(csb, streams);
+			stateHolder.deactivate();
+
+			// Create an independent retrieval
+			rsb = optimizer->generateRetrieval(stream.number, sortPtr, false, false);
+
+			// Create a nested loop join from the priorly processed streams
+			const auto priorRsb = (rsbs.getCount() == 1) ? rsbs[0] :
+				FB_NEW_POOL(getPool()) NestedLoopJoin(csb, rsbs.getCount(), rsbs.begin());
+
+			const River priorRiver(csb, priorRsb, nullptr, streams);
+
+			// Prepare record sources and corresponding equivalence keys for hash-joining
+			RecordSource* hashJoinRsbs[] = {priorRsb, rsb};
+
+			HalfStaticArray<NestValueArray*, OPT_STATIC_ITEMS> keys;
+
+			keys.add(FB_NEW_POOL(getPool()) NestValueArray(getPool()));
+			keys.add(FB_NEW_POOL(getPool()) NestValueArray(getPool()));
+
+			for (const auto match : stream.equiMatches)
+			{
+				NestConst<ValueExprNode> node1;
+				NestConst<ValueExprNode> node2;
+
+				if (!optimizer->getEquiJoinKeys(match, &node1, &node2))
+					fb_assert(false);
+
+				if (!priorRiver.isReferenced(node1))
+				{
+					fb_assert(priorRiver.isReferenced(node2));
+
+					// Swap the sides
+					std::swap(node1, node2);
+				}
+
+				keys[0]->add(node1);
+				keys[1]->add(node2);
+			}
+
+			// Ensure the smallest stream is the one to be hashed
+			if (rsb->getCardinality() > priorRsb->getCardinality())
+			{
+				// Swap the sides
+				std::swap(hashJoinRsbs[0], hashJoinRsbs[1]);
+				std::swap(keys[0], keys[1]);
+			}
+
+			// Create a hash join
+			rsb = FB_NEW_POOL(getPool())
+				HashJoin(tdbb, csb, 2, hashJoinRsbs, keys.begin(), stream.selectivity);
+
+			// Clear priorly processed rsb's, as they're already incorporated into a hash join
+			rsbs.clear();
+		}
+		else
+			rsb = optimizer->generateRetrieval(stream.number, sortPtr, false, false);
+
 		rsbs.add(rsb);
+		streams.add(stream.number);
 		sortPtr = nullptr;
 	}
 
-	const auto rsb = (rsbs.getCount() == 1) ? rsbs[0] :
+	// Create a nested loop join from the processed streams
+	rsb = (rsbs.getCount() == 1) ? rsbs[0] :
 		FB_NEW_POOL(getPool()) NestedLoopJoin(csb, rsbs.getCount(), rsbs.begin());
 
 	// Allocate a river block and move the best order into it
-	const auto river = FB_NEW_POOL(getPool()) River(csb, rsb, nullptr, bestStreams);
+	const auto river = FB_NEW_POOL(getPool()) River(csb, rsb, nullptr, streams);
 	river->deactivate(csb);
 	return river;
 }
@@ -484,21 +608,21 @@ void InnerJoin::getIndexedRelationships(StreamInfo* testStream)
 					  testStream->stream, name.c_str());
 #endif
 
-	const auto tail = &csb->csb_rpt[testStream->stream];
+	const auto tail = &csb->csb_rpt[testStream->number];
 
-	Retrieval retrieval(tdbb, optimizer, testStream->stream, false, false, nullptr, true);
+	Retrieval retrieval(tdbb, optimizer, testStream->number, false, false, nullptr, true);
 	const auto candidate = retrieval.getInversion();
 
 	for (const auto baseStream : innerStreams)
 	{
-		if (baseStream->stream != testStream->stream &&
-			candidate->dependentFromStreams.exist(baseStream->stream))
+		if (baseStream->number != testStream->number &&
+			candidate->dependentFromStreams.exist(baseStream->number))
 		{
 			// If the base stream already depends on the testing stream, don't store it again
 			bool found = false;
 			for (const auto& relationship : baseStream->indexedRelationships)
 			{
-				if (relationship.stream == testStream->stream)
+				if (relationship.stream == testStream->number)
 				{
 					found = true;
 					break;
@@ -515,7 +639,7 @@ void InnerJoin::getIndexedRelationships(StreamInfo* testStream)
 			// with the base stream active as without the base stream
 			// then the test stream has a indexed relationship with the base stream.
 			IndexRelationship indexRelationship;
-			indexRelationship.stream = testStream->stream;
+			indexRelationship.stream = testStream->number;
 			indexRelationship.unique = candidate->unique;
 			indexRelationship.cost = candidate->cost;
 			indexRelationship.cardinality = candidate->unique ?
@@ -540,7 +664,7 @@ InnerJoin::StreamInfo* InnerJoin::getStreamInfo(StreamType stream)
 {
 	for (FB_SIZE_T i = 0; i < innerStreams.getCount(); i++)
 	{
-		if (innerStreams[i]->stream == stream)
+		if (innerStreams[i]->number == stream)
 			return innerStreams[i];
 	}
 
@@ -561,8 +685,8 @@ void InnerJoin::printBestOrder() const
 	const auto end = bestStreams.end();
 	for (auto iter = bestStreams.begin(); iter != end; iter++)
 	{
-		const auto name = optimizer->getStreamName(*iter);
-		optimizer->printf(" %u (%s)", *iter, name.c_str());
+		const auto name = optimizer->getStreamName(iter->number);
+		optimizer->printf(" %u (%s)", iter->number, name.c_str());
 
 		if (iter != end - 1)
 			optimizer->printf(",");
@@ -616,9 +740,9 @@ void InnerJoin::printStartOrder() const
 		const auto innerStream = *iter;
 		if (!innerStream->used)
 		{
-			const auto name = optimizer->getStreamName(innerStream->stream);
+			const auto name = optimizer->getStreamName(innerStream->number);
 			optimizer->printf(" %u (%s) base cost (%1.2f)",
-							  innerStream->stream, name.c_str(), innerStream->baseCost);
+							  innerStream->number, name.c_str(), innerStream->baseCost);
 
 			if (iter != end - 1)
 				optimizer->printf(",");