database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-08-07 03:22:57 +03:00
Code Activity

fix(disk-based-join): MCOL-5626 Fix for race in DJS with outer join. (#3064)

Browse Source
This commit is contained in:
Denis Khalikov
2023-12-15 11:20:27 +03:00
committed by GitHub
parent fb496644f8
commit 74c1a38f2c
2 changed files with 18 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
dbcon/joblist/diskjoinstep.cpp
View File

@@ -515,21 +515,20 @@ void DiskJoinStep::joinFcn(const uint32_t threadID)
if (l_outputRG.getRowCount() == 8192) if (l_outputRG.getRowCount() == 8192)
{ {
outputDL->insert(rgData); outputResult(rgData);
// cout << "inserting a full RG" << endl; // cout << "inserting a full RG" << endl;
if (thjs) if (thjs)
{ {
// FIXME: Possible false positive. Something wrong with this calculation, just put a warning
// until fixed.
if (!thjs->getMemory(l_outputRG.getMaxDataSize())) if (!thjs->getMemory(l_outputRG.getMaxDataSize()))
{ {
// FIXME: This is also looks wrong.
// calculate guess of size required for error message // calculate guess of size required for error message
uint64_t memReqd = (unmatched.size() * outputRG.getDataSize(1)) / 1048576; uint64_t memReqd = (unmatched.size() * outputRG.getDataSize(1)) / 1048576;
Message::Args args; uint64_t memLimit = thjs->resourceManager->getConfiguredUMMemLimit() / 1048576;
args.add(memReqd); std::cerr << "DiskJoin::joinFcn() possible OOM for the join result, mem required: "
args.add(thjs->resourceManager->getConfiguredUMMemLimit() / 1048576); << memReqd << " mem limit: " << memLimit << std::endl;
std::cerr << logging::IDBErrorInfo::instance()->errorMsg(logging::ERR_JOIN_RESULT_TOO_BIG,
args)
<< " @" << __FILE__ << ":" << __LINE__;
throw logging::IDBExcept(logging::ERR_JOIN_RESULT_TOO_BIG, args);
} }
} }
@@ -543,7 +542,7 @@ void DiskJoinStep::joinFcn(const uint32_t threadID)
} }
if (l_outputRG.getRowCount()) if (l_outputRG.getRowCount())
outputResult({rgData}); outputResult(rgData);
if (thjs) if (thjs)
thjs->returnMemory(); thjs->returnMemory();
} }
@@ -629,6 +628,12 @@ void DiskJoinStep::outputResult(const std::vector<rowgroup::RGData>& result)
outputDL->insert(rgData); outputDL->insert(rgData);
} }
void DiskJoinStep::outputResult(const rowgroup::RGData& result)
{
std::lock_guard<std::mutex> lk(outputMutex);
outputDL->insert(result);
}
void DiskJoinStep::spawnJobs(const std::vector<std::vector<JoinPartition*>>& joinPartitionsJobs, void DiskJoinStep::spawnJobs(const std::vector<std::vector<JoinPartition*>>& joinPartitionsJobs,
const uint32_t smallSideSizeLimitPerThread) const uint32_t smallSideSizeLimitPerThread)
{ {

1
dbcon/joblist/diskjoinstep.h
View File

@@ -53,6 +53,7 @@ class DiskJoinStep : public JobStep
void prepareJobs(const std::vector<joiner::JoinPartition*>& joinPartitions, void prepareJobs(const std::vector<joiner::JoinPartition*>& joinPartitions,
JoinPartitionJobs& joinPartitionsJobs); JoinPartitionJobs& joinPartitionsJobs);
void outputResult(const std::vector<rowgroup::RGData>& result); void outputResult(const std::vector<rowgroup::RGData>& result);
void outputResult(const rowgroup::RGData& result);
void spawnJobs(const std::vector<std::vector<joiner::JoinPartition*>>& joinPartitionsJobs, void spawnJobs(const std::vector<std::vector<joiner::JoinPartition*>>& joinPartitionsJobs,
const uint32_t smallSideSizeLimitPerThread); const uint32_t smallSideSizeLimitPerThread);
boost::shared_ptr<joiner::JoinPartition> jp; boost::shared_ptr<joiner::JoinPartition> jp;