Skip to content

Commit fe39f5b

Browse files
yujun777yujun777
authored
[fix](ivm) Answer FE-computable dry runs on the frontend instead of a placeholder backend (#67753)
`REFRESH MATERIALIZED VIEW ... INCREMENTAL WITH DRY RUN` fails with `Unable to resolve host dummy` in cloud mode when the delta is empty. An empty delta plans to a `LogicalEmptyRelation`, so the Nereids planner marks the plan as not needing a backend (`notNeedBackend`) and binds the placeholder `Backend(-1, "dummy", -1)`. Regular queries short-circuit such FE-computable plans inside `handleQueryInFe`; the internal streaming entry used by IVM dry runs (`RefreshMTMVCommand.dryRunRefresh` -> `StmtExecutor.executeInternalQueryAndSend`) skipped that short-circuit and started a coordinator, which then sent fragments to `dummy:-1`. ### What changed - `StmtExecutor.executeInternalQueryCommon` now short-circuits FE-computable plans through `planner.handleQueryInFe` for both the streaming (`sendChannel`) and the collecting internal entry, writing rows to the caller's mysql channel. - `sendResultSet`, `sendMetaData`, `sendTextResultRow` and `sendBinaryResultRow` take an explicit `MysqlChannel` overload; the previous signatures keep using the executor's own channel. - `isHandleQueryInFe` and `ExecutedByFrontend` are set only after the result set was sent, matching the regular query path. - `test_ivm_refresh_dry_run` no longer skips cloud. The new `ivm_dry_run_qt` helper masks `__DORIS_SEQUENCE_COL__` (whose base value differs between cloud and shared-nothing) through the `quickRunTest` row converter, which now optionally receives the result metadata, so both modes share one suite and one `.out`. Both suites cover the empty-delta dry run. Trace issue: #65418
1 parent 5ea9bdc commit fe39f5b

5 files changed

Lines changed: 116 additions & 27 deletions

File tree

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

Lines changed: 61 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1854,11 +1854,16 @@ private void sendMetaData(ResultSetMetaData metaData) throws IOException {
18541854
}
18551855

18561856
private void sendMetaData(ResultSetMetaData metaData, List<FieldInfo> fieldInfos) throws IOException {
1857+
sendMetaData(metaData, fieldInfos, context.getMysqlChannel());
1858+
}
1859+
1860+
private void sendMetaData(ResultSetMetaData metaData, List<FieldInfo> fieldInfos, MysqlChannel channel)
1861+
throws IOException {
18571862
Preconditions.checkState(context.getConnectType() == ConnectType.MYSQL);
18581863
// sends how many columns
18591864
serializer.reset();
18601865
serializer.writeVInt(metaData.getColumnCount());
1861-
context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
1866+
channel.sendOnePacket(serializer.toByteBuffer());
18621867
// send field one by one
18631868
for (int i = 0; i < metaData.getColumns().size(); i++) {
18641869
Column col = metaData.getColumn(i);
@@ -1869,9 +1874,9 @@ private void sendMetaData(ResultSetMetaData metaData, List<FieldInfo> fieldInfos
18691874
} else {
18701875
serializer.writeField(fieldInfos.get(i), col.getType());
18711876
}
1872-
context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
1877+
channel.sendOnePacket(serializer.toByteBuffer());
18731878
}
1874-
sendMetadataTerminatorIfNeeded(context.getMysqlChannel());
1879+
sendMetadataTerminatorIfNeeded(channel);
18751880
}
18761881

18771882
private List<PrimitiveType> exprToStringType(List<Expr> exprs) {
@@ -2034,19 +2039,34 @@ public void sendResultSet(ResultSet resultSet) throws IOException {
20342039
}
20352040

20362041
public void sendResultSet(ResultSet resultSet, List<FieldInfo> fieldInfos) throws IOException {
2042+
sendResultSet(resultSet, fieldInfos, null);
2043+
}
2044+
2045+
/**
2046+
* Sends a FE-computed result set to the given mysql channel. Regular queries use the
2047+
* executor's own channel; internal queries have to stream to the channel of the caller
2048+
* that issued them, because the executor's own channel is not connected to that client.
2049+
*
2050+
* <p>A null channel means the session's own. It is resolved inside the mysql branch on
2051+
* purpose: a connection of any other type has no mysql channel, and asking for one throws,
2052+
* so a caller must be able to hand a result set over without naming a channel first.
2053+
*/
2054+
private void sendResultSet(ResultSet resultSet, List<FieldInfo> fieldInfos, MysqlChannel channel)
2055+
throws IOException {
20372056
if (context.getConnectType().equals(ConnectType.MYSQL)) {
2057+
MysqlChannel targetChannel = channel == null ? context.getMysqlChannel() : channel;
20382058
context.updateReturnRows(resultSet.getResultRows().size());
20392059
// Send meta data.
2040-
sendMetaData(resultSet.getMetaData(), fieldInfos);
2060+
sendMetaData(resultSet.getMetaData(), fieldInfos, targetChannel);
20412061

20422062
// Send result set.
20432063
if (isComStmtExecute) {
20442064
if (LOG.isDebugEnabled()) {
20452065
LOG.debug("Use binary protocol to set result.");
20462066
}
2047-
sendBinaryResultRow(resultSet);
2067+
sendBinaryResultRow(resultSet, targetChannel);
20482068
} else {
2049-
sendTextResultRow(resultSet);
2069+
sendTextResultRow(resultSet, targetChannel);
20502070
}
20512071
context.getState().setEof();
20522072
} else if (context.getConnectType().equals(ConnectType.ARROW_FLIGHT_SQL)) {
@@ -2060,6 +2080,10 @@ public void sendResultSet(ResultSet resultSet, List<FieldInfo> fieldInfos) throw
20602080
}
20612081

20622082
protected void sendTextResultRow(ResultSet resultSet) throws IOException {
2083+
sendTextResultRow(resultSet, context.getMysqlChannel());
2084+
}
2085+
2086+
protected void sendTextResultRow(ResultSet resultSet, MysqlChannel channel) throws IOException {
20632087
for (List<String> row : resultSet.getResultRows()) {
20642088
serializer.reset();
20652089
for (String item : row) {
@@ -2069,11 +2093,15 @@ protected void sendTextResultRow(ResultSet resultSet) throws IOException {
20692093
serializer.writeLenEncodedString(item);
20702094
}
20712095
}
2072-
context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
2096+
channel.sendOnePacket(serializer.toByteBuffer());
20732097
}
20742098
}
20752099

20762100
protected void sendBinaryResultRow(ResultSet resultSet) throws IOException {
2101+
sendBinaryResultRow(resultSet, context.getMysqlChannel());
2102+
}
2103+
2104+
protected void sendBinaryResultRow(ResultSet resultSet, MysqlChannel channel) throws IOException {
20772105
// https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_binary_resultset.html#sect_protocol_binary_resultset_row_value
20782106
ResultSetMetaData metaData = resultSet.getMetaData();
20792107
int nullBitmapLength = (metaData.getColumnCount() + 7 + 2) / 8;
@@ -2136,7 +2164,7 @@ protected void sendBinaryResultRow(ResultSet resultSet) throws IOException {
21362164
}
21372165
}
21382166
}
2139-
context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
2167+
channel.sendOnePacket(serializer.toByteBuffer());
21402168
}
21412169
}
21422170

@@ -2311,6 +2339,31 @@ private List<ResultRow> executeInternalQueryCommon(LogicalPlanAdapter prebuilt,
23112339
planner = new NereidsPlanner(statementContext);
23122340
planner.plan(adapter, context.getSessionVariable().toThrift());
23132341

2342+
// A plan that FE can compute on its own (e.g. the empty delta of a dry run) must be
2343+
// answered by the frontend, like a regular query does. Otherwise the coordinator
2344+
// would send fragments to the placeholder backend registered when no backend is
2345+
// needed (see NereidsPlanner#notNeedBackend), which cannot resolve. Results go to the
2346+
// caller's channel: the executor's own channel is not connected to that client.
2347+
if (context.supportHandleByFe()) {
2348+
Optional<ResultSet> resultSet = planner.handleQueryInFe(adapter);
2349+
if (resultSet.isPresent()) {
2350+
boolean sendToChannel = !collectMode;
2351+
if (sendToChannel) {
2352+
sendResultSet(resultSet.get(), adapter.getFieldInfos(), sendChannel);
2353+
}
2354+
isHandleQueryInFe = true;
2355+
if (context.getSessionVariable().enableProfile() && profile != null) {
2356+
profile.getSummaryProfile().setExecutedByFrontend(true);
2357+
}
2358+
if (sendToChannel) {
2359+
return new ArrayList<>();
2360+
}
2361+
return resultSet.get().getResultRows().stream()
2362+
.map(ResultRow::new)
2363+
.collect(Collectors.toList());
2364+
}
2365+
}
2366+
23142367
if (!collectMode) {
23152368
executeAndSendResult(false, false, adapter, sendChannel, null, null);
23162369
return new ArrayList<>();

‎regression-test/data/mtmv_p0/ivm/test_ivm_refresh_dry_run.out‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,21 +5,25 @@
55
3 1 30
66

77
-- !ivm_dry_run_full --
8-
1 1 1 15 1 1 4097 0
9-
4 4 1 40 1 1 4097 0
8+
1 1 1 15 1 1 [regression-fake-sequence] 0
9+
4 4 1 40 1 1 [regression-fake-sequence] 0
1010

1111
-- !ivm_dry_run_after --
1212
1 1 10
1313
2 1 20
1414
3 1 30
1515

1616
-- !ivm_dry_run_repeat --
17-
1 1 1 15 1 1 4097 0
18-
4 4 1 40 1 1 4097 0
17+
1 1 1 15 1 1 [regression-fake-sequence] 0
18+
4 4 1 40 1 1 [regression-fake-sequence] 0
1919

2020
-- !ivm_dry_run_after_refresh --
2121
1 1 15
2222
2 1 20
2323
3 1 30
2424
4 1 40
2525

26+
-- !ivm_dry_run_empty --
27+
28+
-- !ivm_dry_run_empty_repeat --
29+

‎regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1727,8 +1727,10 @@ class Suite implements GroovyInterceptable {
17271727
return result
17281728
}
17291729

1730-
// rowConverter: { row -> convertedRow }
1730+
// rowConverter: { row -> convertedRow }, or { row, meta -> convertedRow } to inspect the result
1731+
// metadata, for example to mask a column whose value depends on the deployment mode.
17311732
void quickRunTest(String tag, Object arg, boolean isOrder = false, Closure rowConverter = null) {
1733+
boolean converterNeedsMeta = rowConverter != null && rowConverter.maximumNumberOfParameters > 1
17321734
if (context.config.generateOutputFile || context.config.forceGenerateOutputFile) {
17331735
Tuple2<List<List<Object>>, ResultSetMetaData> tupleResult = null
17341736
if (arg instanceof PreparedStatement) {
@@ -1763,7 +1765,9 @@ class Suite implements GroovyInterceptable {
17631765
}
17641766
def (result, meta) = tupleResult
17651767
if (rowConverter != null) {
1766-
result = result.collect { rowConverter.call(it) }
1768+
result = result.collect {
1769+
converterNeedsMeta ? rowConverter.call(it, meta) : rowConverter.call(it)
1770+
}
17671771
}
17681772
if (isOrder) {
17691773
result = sortByToString(result)
@@ -1815,7 +1819,9 @@ class Suite implements GroovyInterceptable {
18151819
}
18161820
def (realResults, meta) = tupleResult
18171821
if (rowConverter != null) {
1818-
realResults = realResults.collect { rowConverter.call(it) }
1822+
realResults = realResults.collect {
1823+
converterNeedsMeta ? rowConverter.call(it, meta) : rowConverter.call(it)
1824+
}
18191825
}
18201826
if (isOrder) {
18211827
realResults = sortByToString(realResults)

‎regression-test/plugins/plugin_planner.groovy‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -135,3 +135,24 @@ Suite.metaClass.explainIvmPlan = { String tag, String sql ->
135135
)
136136
}
137137

138+
// __DORIS_SEQUENCE_COL__ of an IVM dry-run delta encodes the refresh version, whose base value
139+
// differs between cloud and shared-nothing deployments. Mask that column so that both modes can
140+
// share one suite and one .out file.
141+
def maskIvmDryRunSequence = { row, meta ->
142+
for (int i = 1; i <= meta.getColumnCount(); i++) {
143+
if ("__DORIS_SEQUENCE_COL__".equalsIgnoreCase(meta.getColumnLabel(i))) {
144+
def masked = new ArrayList(row)
145+
masked.set(i - 1, "[regression-fake-sequence]")
146+
return masked
147+
}
148+
}
149+
return row
150+
}
151+
152+
// Named ivm_dry_run_qt instead of order_qt_*: Suite.invokeMethod intercepts every method whose
153+
// name starts with qt_ / order_qt_ and derives the tag from the method name, so a metaClass
154+
// method could never be reached under those prefixes.
155+
Suite.metaClass.ivm_dry_run_qt = { String tag, String sql ->
156+
delegate.quickRunTest(tag, sql, true, maskIvmDryRunSequence)
157+
}
158+

‎regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy‎

Lines changed: 17 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -16,14 +16,9 @@
1616
// under the License.
1717

1818
suite("test_ivm_refresh_dry_run") {
19-
// Cloud mode: __DORIS_SEQUENCE_COL__ in the dry-run delta rows derives from cloud txn
20-
// versioning and differs from local (e.g. 6145 vs 4097), so the .out values
21-
// for the sequence column do not apply.
22-
if (isCloudMode()) {
23-
logger.info("skip test_ivm_refresh_dry_run on cloud mode: " +
24-
"__DORIS_SEQUENCE_COL__ differs between cloud and local")
25-
return
26-
}
19+
// __DORIS_SEQUENCE_COL__ of the dry-run delta rows encodes the refresh version, whose base
20+
// value differs between cloud and shared-nothing deployments (e.g. 6145 vs 4097).
21+
// ivm_dry_run_qt masks that column, so cloud and local share this suite and its .out.
2722
sql "DROP MATERIALIZED VIEW IF EXISTS test_ivm_refresh_dry_run_mv"
2823
sql "DROP TABLE IF EXISTS test_ivm_refresh_dry_run_base"
2924

@@ -65,9 +60,9 @@ suite("test_ivm_refresh_dry_run") {
6560

6661
order_qt_ivm_dry_run_before "SELECT k1, cnt, sum_v1 FROM test_ivm_refresh_dry_run_mv"
6762

68-
order_qt_ivm_dry_run_full """
63+
ivm_dry_run_qt("ivm_dry_run_full", """
6964
REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL WITH DRY RUN
70-
"""
65+
""")
7166

7267
// The delta rows picked by a LIMIT depend on scan order (no ORDER BY before the cap),
7368
// so assert only the returned row count instead of exact rows in the .out file.
@@ -77,11 +72,21 @@ suite("test_ivm_refresh_dry_run") {
7772

7873
order_qt_ivm_dry_run_after "SELECT k1, cnt, sum_v1 FROM test_ivm_refresh_dry_run_mv"
7974

80-
order_qt_ivm_dry_run_repeat """
75+
ivm_dry_run_qt("ivm_dry_run_repeat", """
8176
REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL WITH DRY RUN LIMIT 10
82-
"""
77+
""")
8378

8479
sql "REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL"
8580
waitingMTMVTaskFinishedByMvName("test_ivm_refresh_dry_run_mv")
8681
order_qt_ivm_dry_run_after_refresh "SELECT k1, cnt, sum_v1 FROM test_ivm_refresh_dry_run_mv"
82+
83+
// Empty delta: the delta query is a LogicalEmptyRelation and the dry run must return an
84+
// empty result instead of sending fragments to a placeholder backend. Every dry-run
85+
// comparison goes through ivm_dry_run_qt so the sequence column is masked in all of them.
86+
ivm_dry_run_qt("ivm_dry_run_empty", """
87+
REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL WITH DRY RUN
88+
""")
89+
ivm_dry_run_qt("ivm_dry_run_empty_repeat", """
90+
REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL WITH DRY RUN
91+
""")
8792
}

0 commit comments

Comments
 (0)