Skip to content

Commit 96617ab

Browse files
andyfengHKUandyfengHKU
authored
Merge pull request #977 from graphflowdb/prepare-fix
prepare returns logical plan instead oh physical plan
2 parents f99a0bd + 1a6ee26 commit 96617ab

File tree

8 files changed

+43
-46
lines changed

8 files changed

+43
-46
lines changed

‎src/main/connection.cpp

+10-15
Original file line numberDiff line numberDiff line change
@@ -76,14 +76,14 @@ std::unique_ptr<PreparedStatement> Connection::prepareNoLock(const string& query
7676
auto compilingTimer = TimeMetric(true /* enable */);
7777
compilingTimer.start();
7878
unique_ptr<ExecutionContext> executionContext;
79-
unique_ptr<PhysicalPlan> physicalPlan;
79+
unique_ptr<LogicalPlan> logicalPlan;
8080
try {
8181
auto statement = Parser::parseQuery(query);
8282
auto binder = Binder(*database->catalog);
8383
auto boundStatement = binder.bind(*statement);
8484
setQuerySummaryAndPreparedStatement(statement.get(), binder, preparedStatement.get());
8585
// planning
86-
auto logicalPlan = Planner::getBestPlan(*database->catalog,
86+
logicalPlan = Planner::getBestPlan(*database->catalog,
8787
database->getStorageManager()->getNodesStore().getNodesStatisticsAndDeletedIDs(),
8888
database->getStorageManager()->getRelsStore().getRelsStatistics(), *boundStatement);
8989
if (logicalPlan->isDDLOrCopyCSV()) {
@@ -92,17 +92,13 @@ std::unique_ptr<PreparedStatement> Connection::prepareNoLock(const string& query
9292
} else {
9393
preparedStatement->createResultHeader(logicalPlan->getExpressionsToCollect());
9494
}
95-
// mapping
96-
auto mapper = PlanMapper(
97-
*database->storageManager, database->getMemoryManager(), database->catalog.get());
98-
physicalPlan = mapper.mapLogicalPlanToPhysical(move(logicalPlan));
9995
} catch (Exception& exception) {
10096
preparedStatement->success = false;
10197
preparedStatement->errMsg = exception.what();
10298
}
10399
compilingTimer.stop();
104100
preparedStatement->preparedSummary.compilingTime = compilingTimer.getElapsedTimeMS();
105-
preparedStatement->physicalPlan = move(physicalPlan);
101+
preparedStatement->logicalPlan = std::move(logicalPlan);
106102
return preparedStatement;
107103
}
108104

@@ -209,10 +205,7 @@ unique_ptr<QueryResult> Connection::executePlan(unique_ptr<LogicalPlan> logicalP
209205
lock_t lck{mtx};
210206
auto preparedStatement = make_unique<PreparedStatement>();
211207
preparedStatement->createResultHeader(logicalPlan->getExpressionsToCollect());
212-
auto mapper =
213-
PlanMapper(*database->storageManager, database->getMemoryManager(), database->getCatalog());
214-
auto physicalPlan = mapper.mapLogicalPlanToPhysical(move(logicalPlan));
215-
preparedStatement->physicalPlan = move(physicalPlan);
208+
preparedStatement->logicalPlan = std::move(logicalPlan);
216209
return executeAndAutoCommitIfNecessaryNoLock(preparedStatement.get());
217210
}
218211

@@ -250,6 +243,9 @@ void Connection::bindParametersNoLock(
250243

251244
std::unique_ptr<QueryResult> Connection::executeAndAutoCommitIfNecessaryNoLock(
252245
PreparedStatement* preparedStatement) {
246+
auto mapper = PlanMapper(
247+
*database->storageManager, database->getMemoryManager(), database->catalog.get());
248+
auto physicalPlan = mapper.mapLogicalPlanToPhysical(preparedStatement->logicalPlan.get());
253249
auto queryResult = make_unique<QueryResult>(preparedStatement->preparedSummary);
254250
auto profiler = make_unique<Profiler>();
255251
auto executionContext = make_unique<ExecutionContext>(clientContext->numThreadsForExecution,
@@ -287,8 +283,8 @@ std::unique_ptr<QueryResult> Connection::executeAndAutoCommitIfNecessaryNoLock(
287283
"transaction or set the transaction mode of the connection to AUTO_COMMIT");
288284
}
289285
executionContext->transaction = activeTransaction.get();
290-
resultFT = database->queryProcessor->execute(
291-
preparedStatement->physicalPlan.get(), executionContext.get());
286+
resultFT =
287+
database->queryProcessor->execute(physicalPlan.get(), executionContext.get());
292288
if (AUTO_COMMIT == transactionMode) {
293289
commitNoLock();
294290
}
@@ -302,8 +298,7 @@ std::unique_ptr<QueryResult> Connection::executeAndAutoCommitIfNecessaryNoLock(
302298
queryResult->setResultHeaderAndTable(
303299
preparedStatement->resultHeader->copy(), move(resultFT));
304300
}
305-
auto planPrinter =
306-
make_unique<PlanPrinter>(preparedStatement->physicalPlan.get(), move(profiler));
301+
auto planPrinter = make_unique<PlanPrinter>(physicalPlan.get(), move(profiler));
307302
queryResult->querySummary->planInJson = planPrinter->printPlanToJson();
308303
queryResult->querySummary->planInOstream = planPrinter->printPlanToOstream();
309304
return queryResult;

‎src/main/include/prepared_statement.h

+2-2
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ class PreparedStatement {
2828
resultHeader = make_unique<QueryResultHeader>(move(expressions));
2929
}
3030

31-
inline bool isReadOnly() { return physicalPlan->isReadOnly(); }
31+
inline bool isReadOnly() { return logicalPlan->isReadOnly(); }
3232

3333
private:
3434
bool allowActiveTransaction;
@@ -37,7 +37,7 @@ class PreparedStatement {
3737
PreparedSummary preparedSummary;
3838
unordered_map<string, shared_ptr<Literal>> parameterMap;
3939
unique_ptr<QueryResultHeader> resultHeader;
40-
unique_ptr<PhysicalPlan> physicalPlan;
40+
unique_ptr<LogicalPlan> logicalPlan;
4141
};
4242

4343
} // namespace main

‎src/processor/mapper/include/plan_mapper.h

+1-1
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ class PlanMapper {
2424
: storageManager{storageManager}, memoryManager{memoryManager},
2525
expressionMapper{}, catalog{catalog}, physicalOperatorID{0} {}
2626

27-
unique_ptr<PhysicalPlan> mapLogicalPlanToPhysical(unique_ptr<LogicalPlan> logicalPlan);
27+
unique_ptr<PhysicalPlan> mapLogicalPlanToPhysical(LogicalPlan* logicalPlan);
2828

2929
private:
3030
unique_ptr<PhysicalOperator> mapLogicalOperatorToPhysical(

‎src/processor/mapper/plan_mapper.cpp

+1-1
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ using namespace graphflow::planner;
1111
namespace graphflow {
1212
namespace processor {
1313

14-
unique_ptr<PhysicalPlan> PlanMapper::mapLogicalPlanToPhysical(unique_ptr<LogicalPlan> logicalPlan) {
14+
unique_ptr<PhysicalPlan> PlanMapper::mapLogicalPlanToPhysical(LogicalPlan* logicalPlan) {
1515
auto mapperContext = MapperContext(make_unique<ResultSetDescriptor>(*logicalPlan->getSchema()));
1616
auto prevOperator = mapLogicalOperatorToPhysical(logicalPlan->getLastOperator(), mapperContext);
1717
auto lastOperator = appendResultCollector(logicalPlan->getExpressionsToCollect(),

‎test/main/prepare_test.cpp

+8-3
Original file line numberDiff line numberDiff line change
@@ -114,7 +114,12 @@ TEST_F(ApiTest, ParamTypeError) {
114114
}
115115

116116
TEST_F(ApiTest, MultipleExecutionOfPreparedStatement) {
117-
auto preparedStatement = conn->prepare("MATCH (o:organisation) RETURN 2 + $a");
118-
ASSERT_TRUE(conn->execute(preparedStatement.get(), make_pair(string("a"), (int64_t)1)));
119-
ASSERT_TRUE(conn->execute(preparedStatement.get(), make_pair(string("a"), (int64_t)2)));
117+
auto preparedStatement =
118+
conn->prepare("MATCH (a:person) WHERE a.fName STARTS WITH $n RETURN a.ID, a.fName");
119+
auto result = conn->execute(preparedStatement.get(), make_pair(string("n"), "A"));
120+
auto groundTruth = vector<string>{"0|Alice"};
121+
ASSERT_EQ(groundTruth, TestHelper::convertResultToString(*result));
122+
result = conn->execute(preparedStatement.get(), make_pair(string("n"), "B"));
123+
groundTruth = vector<string>{"2|Bob"};
124+
ASSERT_EQ(groundTruth, TestHelper::convertResultToString(*result));
120125
}

‎test/runner/e2e_copy_csv_transaction_test.cpp

+12-12
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@
22

33
#include "test/test_utility/include/test_helper.h"
44

5+
#include "src/processor/mapper/include/plan_mapper.h"
6+
57
using namespace graphflow::testing;
68

79
namespace graphflow {
@@ -15,10 +17,8 @@ class TinySnbCopyCSVTransactionTest : public EmptyDBTest {
1517
createDBAndConn();
1618
catalog = conn->database->getCatalog();
1719
profiler = make_unique<Profiler>();
18-
bufferManager = make_unique<BufferManager>();
19-
memoryManager = make_unique<MemoryManager>(bufferManager.get());
20-
executionContext = make_unique<ExecutionContext>(
21-
1 /* numThreads */, profiler.get(), memoryManager.get(), bufferManager.get());
20+
executionContext = make_unique<ExecutionContext>(1 /* numThreads */, profiler.get(),
21+
database->getMemoryManager(), database->getBufferManager());
2222
}
2323

2424
void initWithoutLoadingGraph() {
@@ -76,8 +76,10 @@ class TinySnbCopyCSVTransactionTest : public EmptyDBTest {
7676
conn->query(createPersonTableCMD);
7777
auto preparedStatement = conn->prepareNoLock(copyPersonTableCMD);
7878
conn->beginTransactionNoLock(WRITE);
79-
database->queryProcessor->execute(
80-
preparedStatement->physicalPlan.get(), executionContext.get());
79+
auto mapper = PlanMapper(
80+
*database->storageManager, database->getMemoryManager(), database->catalog.get());
81+
auto physicalPlan = mapper.mapLogicalPlanToPhysical(preparedStatement->logicalPlan.get());
82+
database->queryProcessor->execute(physicalPlan.get(), executionContext.get());
8183
auto tableID = catalog->getReadOnlyVersion()->getNodeTableIDFromName("person");
8284
validateDatabaseStateBeforeCheckPointCopyNodeCSV(tableID);
8385
if (transactionTestType == TransactionTestType::RECOVERY) {
@@ -156,10 +158,10 @@ class TinySnbCopyCSVTransactionTest : public EmptyDBTest {
156158
conn->query(createKnowsTableCMD);
157159
auto preparedStatement = conn->prepareNoLock(copyKnowsTableCMD);
158160
conn->beginTransactionNoLock(WRITE);
159-
auto profiler = make_unique<Profiler>();
160-
auto bufferManager = make_unique<BufferManager>();
161-
database->queryProcessor->execute(
162-
preparedStatement->physicalPlan.get(), executionContext.get());
161+
auto mapper = PlanMapper(
162+
*database->storageManager, database->getMemoryManager(), database->catalog.get());
163+
auto physicalPlan = mapper.mapLogicalPlanToPhysical(preparedStatement->logicalPlan.get());
164+
database->queryProcessor->execute(physicalPlan.get(), executionContext.get());
163165
auto tableID = catalog->getReadOnlyVersion()->getRelTableIDFromName("knows");
164166
validateDatabaseStateBeforeCheckPointCopyRelCSV(tableID);
165167
if (transactionTestType == TransactionTestType::RECOVERY) {
@@ -186,8 +188,6 @@ class TinySnbCopyCSVTransactionTest : public EmptyDBTest {
186188
"validInterval INTERVAL, comments STRING[], MANY_MANY)";
187189
string copyKnowsTableCMD = "COPY knows FROM \"dataset/tinysnb/eKnows.csv\"";
188190
unique_ptr<Profiler> profiler;
189-
unique_ptr<BufferManager> bufferManager;
190-
unique_ptr<MemoryManager> memoryManager;
191191
unique_ptr<ExecutionContext> executionContext;
192192
};
193193
} // namespace transaction

‎test/runner/e2e_ddl_test.cpp

+8-8
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
#include "test/test_utility/include/test_helper.h"
22

3+
#include "src/processor/mapper/include/plan_mapper.h"
4+
35
using namespace graphflow::testing;
46

57
namespace graphflow {
@@ -90,10 +92,8 @@ class TinySnbDDLTest : public DBTest {
9092
DBTest::SetUp();
9193
catalog = conn->database->getCatalog();
9294
profiler = make_unique<Profiler>();
93-
bufferManager = make_unique<BufferManager>();
94-
memoryManager = make_unique<MemoryManager>(bufferManager.get());
95-
executionContext = make_unique<ExecutionContext>(
96-
1 /* numThreads */, profiler.get(), memoryManager.get(), bufferManager.get());
95+
executionContext = make_unique<ExecutionContext>(1 /* numThreads */, profiler.get(),
96+
database->getMemoryManager(), database->getBufferManager());
9797
}
9898

9999
void initWithoutLoadingGraph() {
@@ -215,15 +215,15 @@ class TinySnbDDLTest : public DBTest {
215215
void executeDDLWithoutCommit(string query) {
216216
auto preparedStatement = conn->prepareNoLock(query);
217217
conn->beginTransactionNoLock(WRITE);
218-
database->queryProcessor->execute(
219-
preparedStatement->physicalPlan.get(), executionContext.get());
218+
auto mapper = PlanMapper(
219+
*database->storageManager, database->getMemoryManager(), database->catalog.get());
220+
auto physicalPlan = mapper.mapLogicalPlanToPhysical(preparedStatement->logicalPlan.get());
221+
database->queryProcessor->execute(physicalPlan.get(), executionContext.get());
220222
}
221223

222224
Catalog* catalog;
223225
unique_ptr<ExecutionContext> executionContext;
224226
unique_ptr<Profiler> profiler;
225-
unique_ptr<BufferManager> bufferManager;
226-
unique_ptr<MemoryManager> memoryManager;
227227
};
228228
} // namespace transaction
229229
} // namespace graphflow

‎tools/join_order_pick/jo_connection.cpp

+1-4
Original file line numberDiff line numberDiff line change
@@ -37,10 +37,7 @@ unique_ptr<QueryResult> JOConnection::query(const string& query, const string& e
3737
}
3838
}
3939
preparedStatement->createResultHeader(logicalPlan->getExpressionsToCollect());
40-
// mapping
41-
auto mapper = PlanMapper(
42-
*database->storageManager, database->getMemoryManager(), database->catalog.get());
43-
preparedStatement->physicalPlan = mapper.mapLogicalPlanToPhysical(move(logicalPlan));
40+
preparedStatement->logicalPlan = std::move(logicalPlan);
4441
return executeAndAutoCommitIfNecessaryNoLock(preparedStatement.get());
4542
} catch (Exception& exception) {
4643
string errMsg = exception.what();

0 commit comments

Comments
 (0)