Skip to content

Commit dc80ecf

Browse files
924060929924060929
authored
[fix](Nereids) fix bug of dry run query with sql cache (#33799)
1. dry run query should not use sql cache 2. fix test sql cache in cloud mode 3. enable cache OneRowRelation and EmptyRelation in frontend to skip parse sql
1 parent ce2c1ad commit dc80ecf

8 files changed

Lines changed: 241 additions & 36 deletions

File tree

‎fe/fe-core/src/main/java/org/apache/doris/common/NereidsSqlCacheManager.java‎

Lines changed: 35 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@
5151
import org.apache.doris.proto.InternalService;
5252
import org.apache.doris.proto.Types.PUniqueId;
5353
import org.apache.doris.qe.ConnectContext;
54+
import org.apache.doris.qe.ResultSet;
5455
import org.apache.doris.qe.cache.CacheAnalyzer;
5556
import org.apache.doris.qe.cache.SqlCache;
5657

@@ -96,8 +97,8 @@ public static synchronized void updateConfig() {
9697
}
9798

9899
private static Cache<String, SqlCacheContext> buildSqlCaches(int sqlCacheNum, long cacheIntervalSeconds) {
99-
sqlCacheNum = sqlCacheNum <= 0 ? 100 : sqlCacheNum;
100-
cacheIntervalSeconds = cacheIntervalSeconds <= 0 ? 30 : cacheIntervalSeconds;
100+
sqlCacheNum = sqlCacheNum < 0 ? 100 : sqlCacheNum;
101+
cacheIntervalSeconds = cacheIntervalSeconds < 0 ? 30 : cacheIntervalSeconds;
101102

102103
return Caffeine.newBuilder()
103104
.maximumSize(sqlCacheNum)
@@ -107,6 +108,22 @@ private static Cache<String, SqlCacheContext> buildSqlCaches(int sqlCacheNum, lo
107108
.build();
108109
}
109110

111+
/** tryAddFeCache */
112+
public void tryAddFeSqlCache(ConnectContext connectContext, String sql) {
113+
Optional<SqlCacheContext> sqlCacheContextOpt = connectContext.getStatementContext().getSqlCacheContext();
114+
if (!sqlCacheContextOpt.isPresent()) {
115+
return;
116+
}
117+
118+
SqlCacheContext sqlCacheContext = sqlCacheContextOpt.get();
119+
UserIdentity currentUserIdentity = connectContext.getCurrentUserIdentity();
120+
String key = currentUserIdentity.toString() + ":" + sql.trim();
121+
if ((sqlCaches.getIfPresent(key) == null) && sqlCacheContext.getOrComputeCacheKeyMd5() != null
122+
&& sqlCacheContext.getResultSetInFe().isPresent()) {
123+
sqlCaches.put(key, sqlCacheContext);
124+
}
125+
}
126+
110127
/** tryAddCache */
111128
public void tryAddCache(
112129
ConnectContext connectContext, String sql,
@@ -178,6 +195,19 @@ public Optional<LogicalSqlCache> tryParseSql(ConnectContext connectContext, Stri
178195
}
179196

180197
try {
198+
Optional<ResultSet> resultSetInFe = sqlCacheContext.getResultSetInFe();
199+
if (resultSetInFe.isPresent()) {
200+
MetricRepo.COUNTER_CACHE_HIT_SQL.increase(1L);
201+
202+
String cachedPlan = sqlCacheContext.getPhysicalPlan();
203+
LogicalSqlCache logicalSqlCache = new LogicalSqlCache(
204+
sqlCacheContext.getQueryId(), sqlCacheContext.getColLabels(),
205+
sqlCacheContext.getResultExprs(), resultSetInFe, ImmutableList.of(),
206+
"none", cachedPlan
207+
);
208+
return Optional.of(logicalSqlCache);
209+
}
210+
181211
Status status = new Status();
182212
PUniqueId cacheKeyMd5 = sqlCacheContext.getOrComputeCacheKeyMd5();
183213
InternalService.PFetchCacheResult cacheData =
@@ -195,7 +225,9 @@ public Optional<LogicalSqlCache> tryParseSql(ConnectContext connectContext, Stri
195225

196226
LogicalSqlCache logicalSqlCache = new LogicalSqlCache(
197227
sqlCacheContext.getQueryId(), sqlCacheContext.getColLabels(),
198-
sqlCacheContext.getResultExprs(), cacheValues, backendAddress, cachedPlan);
228+
sqlCacheContext.getResultExprs(), Optional.empty(),
229+
cacheValues, backendAddress, cachedPlan
230+
);
199231
return Optional.of(logicalSqlCache);
200232
}
201233
return invalidateCache(key);

‎fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java‎

Lines changed: 61 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import org.apache.doris.analysis.LiteralExpr;
2323
import org.apache.doris.analysis.StatementBase;
2424
import org.apache.doris.catalog.Column;
25+
import org.apache.doris.catalog.Env;
2526
import org.apache.doris.catalog.MTMV;
2627
import org.apache.doris.common.NereidsException;
2728
import org.apache.doris.common.Pair;
@@ -54,6 +55,7 @@
5455
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
5556
import org.apache.doris.nereids.trees.plans.logical.LogicalSqlCache;
5657
import org.apache.doris.nereids.trees.plans.physical.PhysicalCatalogRelation;
58+
import org.apache.doris.nereids.trees.plans.physical.PhysicalEmptyRelation;
5759
import org.apache.doris.nereids.trees.plans.physical.PhysicalOneRowRelation;
5860
import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
5961
import org.apache.doris.nereids.trees.plans.physical.PhysicalResultSink;
@@ -66,8 +68,10 @@
6668
import org.apache.doris.qe.ConnectContext;
6769
import org.apache.doris.qe.ResultSet;
6870
import org.apache.doris.qe.ResultSetMetaData;
71+
import org.apache.doris.qe.cache.CacheAnalyzer;
6972

7073
import com.google.common.annotations.VisibleForTesting;
74+
import com.google.common.collect.ImmutableList;
7175
import com.google.common.collect.Lists;
7276
import org.apache.logging.log4j.LogManager;
7377
import org.apache.logging.log4j.Logger;
@@ -168,8 +172,9 @@ public Plan plan(LogicalPlan plan, PhysicalProperties requireProperties,
168172
LogicalSqlCache logicalSqlCache = (LogicalSqlCache) plan;
169173
physicalPlan = new PhysicalSqlCache(
170174
logicalSqlCache.getQueryId(), logicalSqlCache.getColumnLabels(),
171-
logicalSqlCache.getResultExprs(), logicalSqlCache.getCacheValues(),
172-
logicalSqlCache.getBackendAddress(), logicalSqlCache.getPlanBody()
175+
logicalSqlCache.getResultExprs(), logicalSqlCache.getResultSetInFe(),
176+
logicalSqlCache.getCacheValues(), logicalSqlCache.getBackendAddress(),
177+
logicalSqlCache.getPlanBody()
173178
);
174179
return physicalPlan;
175180
}
@@ -528,31 +533,66 @@ public Optional<ResultSet> handleQueryInFe(StatementBase parsedStmt) {
528533
if (!(parsedStmt instanceof LogicalPlanAdapter)) {
529534
return Optional.empty();
530535
}
531-
if (!(physicalPlan instanceof PhysicalResultSink)) {
532-
return Optional.empty();
536+
if (physicalPlan instanceof PhysicalSqlCache
537+
&& ((PhysicalSqlCache) physicalPlan).getResultSet().isPresent()) {
538+
return Optional.of(((PhysicalSqlCache) physicalPlan).getResultSet().get());
533539
}
534-
if (!(((PhysicalResultSink<?>) physicalPlan).child() instanceof PhysicalOneRowRelation)) {
540+
if (!(physicalPlan instanceof PhysicalResultSink)) {
535541
return Optional.empty();
536542
}
537-
PhysicalOneRowRelation physicalOneRowRelation
538-
= (PhysicalOneRowRelation) ((PhysicalResultSink<?>) physicalPlan).child();
539-
List<Column> columns = Lists.newArrayList();
540-
List<String> data = Lists.newArrayList();
541-
for (int i = 0; i < physicalOneRowRelation.getProjects().size(); i++) {
542-
NamedExpression item = physicalOneRowRelation.getProjects().get(i);
543-
NamedExpression output = physicalPlan.getOutput().get(i);
544-
Expression expr = item.child(0);
545-
if (expr instanceof Literal) {
546-
LiteralExpr legacyExpr = ((Literal) expr).toLegacyLiteral();
543+
544+
Optional<SqlCacheContext> sqlCacheContext = statementContext.getSqlCacheContext();
545+
boolean enableSqlCache
546+
= CacheAnalyzer.canUseSqlCache(statementContext.getConnectContext().getSessionVariable());
547+
Plan child = physicalPlan.child(0);
548+
if (child instanceof PhysicalOneRowRelation) {
549+
PhysicalOneRowRelation physicalOneRowRelation = (PhysicalOneRowRelation) physicalPlan.child(0);
550+
List<Column> columns = Lists.newArrayList();
551+
List<String> data = Lists.newArrayList();
552+
for (int i = 0; i < physicalOneRowRelation.getProjects().size(); i++) {
553+
NamedExpression item = physicalOneRowRelation.getProjects().get(i);
554+
NamedExpression output = physicalPlan.getOutput().get(i);
555+
Expression expr = item.child(0);
556+
if (expr instanceof Literal) {
557+
LiteralExpr legacyExpr = ((Literal) expr).toLegacyLiteral();
558+
columns.add(new Column(output.getName(), output.getDataType().toCatalogDataType()));
559+
data.add(legacyExpr.getStringValueInFe());
560+
} else {
561+
return Optional.empty();
562+
}
563+
}
564+
565+
ResultSetMetaData metadata = new CommonResultSet.CommonResultSetMetaData(columns);
566+
ResultSet resultSet = new CommonResultSet(metadata, Collections.singletonList(data));
567+
if (sqlCacheContext.isPresent() && enableSqlCache) {
568+
sqlCacheContext.get().setResultSetInFe(resultSet);
569+
Env.getCurrentEnv().getSqlCacheManager().tryAddFeSqlCache(
570+
statementContext.getConnectContext(),
571+
statementContext.getOriginStatement().originStmt
572+
);
573+
}
574+
return Optional.of(resultSet);
575+
} else if (child instanceof PhysicalEmptyRelation) {
576+
List<Column> columns = Lists.newArrayList();
577+
PhysicalEmptyRelation physicalEmptyRelation = (PhysicalEmptyRelation) physicalPlan.child(0);
578+
for (int i = 0; i < physicalEmptyRelation.getProjects().size(); i++) {
579+
NamedExpression output = physicalPlan.getOutput().get(i);
547580
columns.add(new Column(output.getName(), output.getDataType().toCatalogDataType()));
548-
data.add(legacyExpr.getStringValueInFe());
549-
} else {
550-
return Optional.empty();
551581
}
582+
583+
ResultSetMetaData metadata = new CommonResultSet.CommonResultSetMetaData(columns);
584+
ResultSet resultSet = new CommonResultSet(metadata, ImmutableList.of());
585+
if (sqlCacheContext.isPresent() && enableSqlCache) {
586+
sqlCacheContext.get().setResultSetInFe(resultSet);
587+
Env.getCurrentEnv().getSqlCacheManager().tryAddFeSqlCache(
588+
statementContext.getConnectContext(),
589+
statementContext.getOriginStatement().originStmt
590+
);
591+
}
592+
return Optional.of(resultSet);
593+
} else {
594+
return Optional.empty();
552595
}
553-
ResultSetMetaData metadata = new CommonResultSet.CommonResultSetMetaData(columns);
554-
ResultSet resultSet = new CommonResultSet(metadata, Collections.singletonList(data));
555-
return Optional.of(resultSet);
556596
}
557597

558598
@VisibleForTesting

‎fe/fe-core/src/main/java/org/apache/doris/nereids/SqlCacheContext.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import org.apache.doris.nereids.trees.expressions.Variable;
3030
import org.apache.doris.nereids.util.Utils;
3131
import org.apache.doris.proto.Types.PUniqueId;
32+
import org.apache.doris.qe.ResultSet;
3233
import org.apache.doris.qe.cache.CacheProxy;
3334
import org.apache.doris.thrift.TUniqueId;
3435

@@ -83,6 +84,7 @@ public class SqlCacheContext {
8384
private volatile List<String> colLabels;
8485

8586
private volatile PUniqueId cacheKeyMd5;
87+
private volatile ResultSet resultSetInFe;
8688

8789
public SqlCacheContext(UserIdentity userIdentity, TUniqueId queryId) {
8890
this.userIdentity = Objects.requireNonNull(userIdentity, "userIdentity cannot be null");
@@ -378,6 +380,14 @@ public void setOriginSql(String originSql) {
378380
this.originSql = originSql.trim();
379381
}
380382

383+
public Optional<ResultSet> getResultSetInFe() {
384+
return Optional.ofNullable(resultSetInFe);
385+
}
386+
387+
public void setResultSetInFe(ResultSet resultSetInFe) {
388+
this.resultSetInFe = resultSetInFe;
389+
}
390+
381391
/** FullTableName */
382392
@lombok.Data
383393
@lombok.AllArgsConstructor

‎fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalSqlCache.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
3333
import org.apache.doris.nereids.util.Utils;
3434
import org.apache.doris.proto.InternalService;
35+
import org.apache.doris.qe.ResultSet;
3536
import org.apache.doris.thrift.TUniqueId;
3637

3738
import com.google.common.collect.ImmutableList;
@@ -46,18 +47,21 @@ public class LogicalSqlCache extends LogicalLeaf implements SqlCache, TreeString
4647
private final TUniqueId queryId;
4748
private final List<String> columnLabels;
4849
private final List<Expr> resultExprs;
50+
private final Optional<ResultSet> resultSetInFe;
4951
private final List<InternalService.PCacheValue> cacheValues;
5052
private final String backendAddress;
5153
private final String planBody;
5254

5355
/** LogicalSqlCache */
5456
public LogicalSqlCache(TUniqueId queryId,
5557
List<String> columnLabels, List<Expr> resultExprs,
56-
List<InternalService.PCacheValue> cacheValues, String backendAddress, String planBody) {
58+
Optional<ResultSet> resultSetInFe, List<InternalService.PCacheValue> cacheValues,
59+
String backendAddress, String planBody) {
5760
super(PlanType.LOGICAL_SQL_CACHE, Optional.empty(), Optional.empty());
5861
this.queryId = Objects.requireNonNull(queryId, "queryId can not be null");
5962
this.columnLabels = Objects.requireNonNull(columnLabels, "columnLabels can not be null");
6063
this.resultExprs = Objects.requireNonNull(resultExprs, "resultExprs can not be null");
64+
this.resultSetInFe = Objects.requireNonNull(resultSetInFe, "resultSetInFe can not be null");
6165
this.cacheValues = Objects.requireNonNull(cacheValues, "cacheValues can not be null");
6266
this.backendAddress = Objects.requireNonNull(backendAddress, "backendAddress can not be null");
6367
this.planBody = Objects.requireNonNull(planBody, "planBody can not be null");
@@ -67,6 +71,10 @@ public TUniqueId getQueryId() {
6771
return queryId;
6872
}
6973

74+
public Optional<ResultSet> getResultSetInFe() {
75+
return resultSetInFe;
76+
}
77+
7078
public List<InternalService.PCacheValue> getCacheValues() {
7179
return cacheValues;
7280
}

‎fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalSqlCache.java‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@
3232
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
3333
import org.apache.doris.nereids.util.Utils;
3434
import org.apache.doris.proto.InternalService;
35+
import org.apache.doris.proto.InternalService.PCacheValue;
36+
import org.apache.doris.qe.ResultSet;
3537
import org.apache.doris.statistics.Statistics;
3638
import org.apache.doris.thrift.TUniqueId;
3739

@@ -46,19 +48,22 @@ public class PhysicalSqlCache extends PhysicalLeaf implements SqlCache, TreeStri
4648
private final TUniqueId queryId;
4749
private final List<String> columnLabels;
4850
private final List<Expr> resultExprs;
51+
private final Optional<ResultSet> resultSet;
4952
private final List<InternalService.PCacheValue> cacheValues;
5053
private final String backendAddress;
5154
private final String planBody;
5255

5356
/** PhysicalSqlCache */
5457
public PhysicalSqlCache(TUniqueId queryId,
5558
List<String> columnLabels, List<Expr> resultExprs,
56-
List<InternalService.PCacheValue> cacheValues, String backendAddress, String planBody) {
59+
Optional<ResultSet> resultSet, List<InternalService.PCacheValue> cacheValues,
60+
String backendAddress, String planBody) {
5761
super(PlanType.PHYSICAL_SQL_CACHE, Optional.empty(),
5862
new LogicalProperties(() -> ImmutableList.of(), () -> FunctionalDependencies.EMPTY_FUNC_DEPS));
5963
this.queryId = Objects.requireNonNull(queryId, "queryId can not be null");
6064
this.columnLabels = Objects.requireNonNull(columnLabels, "colNames can not be null");
6165
this.resultExprs = Objects.requireNonNull(resultExprs, "resultExprs can not be null");
66+
this.resultSet = Objects.requireNonNull(resultSet, "resultSet can not be null");
6267
this.cacheValues = Objects.requireNonNull(cacheValues, "cacheValues can not be null");
6368
this.backendAddress = Objects.requireNonNull(backendAddress, "backendAddress can not be null");
6469
this.planBody = Objects.requireNonNull(planBody, "planBody can not be null");
@@ -68,6 +73,10 @@ public TUniqueId getQueryId() {
6873
return queryId;
6974
}
7075

76+
public Optional<ResultSet> getResultSet() {
77+
return resultSet;
78+
}
79+
7180
public List<InternalService.PCacheValue> getCacheValues() {
7281
return cacheValues;
7382
}
@@ -90,9 +99,18 @@ public String getPlanBody() {
9099

91100
@Override
92101
public String toString() {
102+
long rowCount = 0;
103+
if (resultSet.isPresent()) {
104+
rowCount = resultSet.get().getResultRows().size();
105+
} else {
106+
for (PCacheValue cacheValue : cacheValues) {
107+
rowCount += cacheValue.getRowsCount();
108+
}
109+
}
93110
return Utils.toSqlString("PhysicalSqlCache[" + id.asInt() + "]",
94111
"queryId", DebugUtil.printId(queryId),
95-
"backend", backendAddress
112+
"backend", backendAddress,
113+
"rowCount", rowCount
96114
);
97115
}
98116

‎fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1661,8 +1661,7 @@ private boolean sendCachedValues(MysqlChannel channel, List<InternalService.PCac
16611661
/**
16621662
* Handle the SelectStmt via Cache.
16631663
*/
1664-
private void handleCacheStmt(CacheAnalyzer cacheAnalyzer, MysqlChannel channel)
1665-
throws Exception {
1664+
private void handleCacheStmt(CacheAnalyzer cacheAnalyzer, MysqlChannel channel) throws Exception {
16661665
InternalService.PFetchCacheResult cacheResult = null;
16671666
boolean wantToParseSqlForSqlCache = planner instanceof NereidsPlanner
16681667
&& CacheAnalyzer.canUseSqlCache(context.getSessionVariable());
@@ -1838,6 +1837,7 @@ private void sendResult(boolean isOutfileQuery, boolean isSendFields, Queriable
18381837
return;
18391838
}
18401839

1840+
boolean isDryRun = ConnectContext.get() != null && ConnectContext.get().getSessionVariable().dryRunQuery;
18411841
while (true) {
18421842
// register the fetch result time.
18431843
profile.getSummaryProfile().setTempStartTime();
@@ -1846,7 +1846,7 @@ private void sendResult(boolean isOutfileQuery, boolean isSendFields, Queriable
18461846

18471847
// for outfile query, there will be only one empty batch send back with eos flag
18481848
// call `copyRowBatch()` first, because batch.getBatch() may be null, if result set is empty
1849-
if (cacheAnalyzer != null && !isOutfileQuery) {
1849+
if (cacheAnalyzer != null && !isOutfileQuery && !isDryRun) {
18501850
cacheAnalyzer.copyRowBatch(batch);
18511851
}
18521852
if (batch.getBatch() != null) {
@@ -1877,10 +1877,10 @@ private void sendResult(boolean isOutfileQuery, boolean isSendFields, Queriable
18771877
break;
18781878
}
18791879
}
1880-
if (cacheAnalyzer != null) {
1880+
if (cacheAnalyzer != null && !isDryRun) {
18811881
if (cacheResult != null && cacheAnalyzer.getHitRange() == Cache.HitRange.Right) {
18821882
isSendFields =
1883-
sendCachedValues(channel, cacheResult.getValuesList(), (Queriable) queryStmt, isSendFields,
1883+
sendCachedValues(channel, cacheResult.getValuesList(), queryStmt, isSendFields,
18841884
false);
18851885
}
18861886

@@ -1898,7 +1898,7 @@ private void sendResult(boolean isOutfileQuery, boolean isSendFields, Queriable
18981898
}
18991899
if (!isSendFields) {
19001900
if (!isOutfileQuery) {
1901-
if (ConnectContext.get() != null && ConnectContext.get().getSessionVariable().dryRunQuery) {
1901+
if (ConnectContext.get() != null && isDryRun) {
19021902
// Return a one row one column result set, with the real result number
19031903
List<String> data = Lists.newArrayList(batch.getQueryStatistics() == null ? "0"
19041904
: batch.getQueryStatistics().getReturnedRows() + "");

‎fe/fe-core/src/main/java/org/apache/doris/qe/cache/CacheAnalyzer.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -215,7 +215,8 @@ public static boolean canUseSqlCache(SessionVariable sessionVariable) {
215215
}
216216

217217
public static boolean commonCacheCondition(SessionVariable sessionVariable) {
218-
return sessionVariable.getSqlSelectLimit() < 0 && sessionVariable.getDefaultOrderByLimit() < 0;
218+
return sessionVariable.getSqlSelectLimit() < 0 && sessionVariable.getDefaultOrderByLimit() < 0
219+
&& !sessionVariable.dryRunQuery;
219220
}
220221

221222
/**

0 commit comments

Comments
 (0)