This is an automated email from the ASF dual-hosted git repository.
yujun777 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new fe39f5b6a42 [fix](ivm) Answer FE-computable dry runs on the frontend
instead of a placeholder backend (#67753)
fe39f5b6a42 is described below
commit fe39f5b6a42de75359358735db32b7020b6d7abe
Author: yujun <[email protected]>
AuthorDate: Fri Sep 11 17:21:05 2026 +0800
[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: https://github.com/apache/doris/issues/65418
---
.../java/org/apache/doris/qe/StmtExecutor.java | 69 +++++++++++++++++++---
.../data/mtmv_p0/ivm/test_ivm_refresh_dry_run.out | 12 ++--
.../org/apache/doris/regression/suite/Suite.groovy | 12 +++-
regression-test/plugins/plugin_planner.groovy | 21 +++++++
.../mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy | 29 +++++----
5 files changed, 116 insertions(+), 27 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
index 701f213c6ef..ef0c28cbeb5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
@@ -1854,11 +1854,16 @@ public class StmtExecutor {
}
private void sendMetaData(ResultSetMetaData metaData, List<FieldInfo>
fieldInfos) throws IOException {
+ sendMetaData(metaData, fieldInfos, context.getMysqlChannel());
+ }
+
+ private void sendMetaData(ResultSetMetaData metaData, List<FieldInfo>
fieldInfos, MysqlChannel channel)
+ throws IOException {
Preconditions.checkState(context.getConnectType() ==
ConnectType.MYSQL);
// sends how many columns
serializer.reset();
serializer.writeVInt(metaData.getColumnCount());
- context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
+ channel.sendOnePacket(serializer.toByteBuffer());
// send field one by one
for (int i = 0; i < metaData.getColumns().size(); i++) {
Column col = metaData.getColumn(i);
@@ -1869,9 +1874,9 @@ public class StmtExecutor {
} else {
serializer.writeField(fieldInfos.get(i), col.getType());
}
- context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
+ channel.sendOnePacket(serializer.toByteBuffer());
}
- sendMetadataTerminatorIfNeeded(context.getMysqlChannel());
+ sendMetadataTerminatorIfNeeded(channel);
}
private List<PrimitiveType> exprToStringType(List<Expr> exprs) {
@@ -2034,19 +2039,34 @@ public class StmtExecutor {
}
public void sendResultSet(ResultSet resultSet, List<FieldInfo> fieldInfos)
throws IOException {
+ sendResultSet(resultSet, fieldInfos, null);
+ }
+
+ /**
+ * Sends a FE-computed result set to the given mysql channel. Regular
queries use the
+ * executor's own channel; internal queries have to stream to the channel
of the caller
+ * that issued them, because the executor's own channel is not connected
to that client.
+ *
+ * <p>A null channel means the session's own. It is resolved inside the
mysql branch on
+ * purpose: a connection of any other type has no mysql channel, and
asking for one throws,
+ * so a caller must be able to hand a result set over without naming a
channel first.
+ */
+ private void sendResultSet(ResultSet resultSet, List<FieldInfo>
fieldInfos, MysqlChannel channel)
+ throws IOException {
if (context.getConnectType().equals(ConnectType.MYSQL)) {
+ MysqlChannel targetChannel = channel == null ?
context.getMysqlChannel() : channel;
context.updateReturnRows(resultSet.getResultRows().size());
// Send meta data.
- sendMetaData(resultSet.getMetaData(), fieldInfos);
+ sendMetaData(resultSet.getMetaData(), fieldInfos, targetChannel);
// Send result set.
if (isComStmtExecute) {
if (LOG.isDebugEnabled()) {
LOG.debug("Use binary protocol to set result.");
}
- sendBinaryResultRow(resultSet);
+ sendBinaryResultRow(resultSet, targetChannel);
} else {
- sendTextResultRow(resultSet);
+ sendTextResultRow(resultSet, targetChannel);
}
context.getState().setEof();
} else if
(context.getConnectType().equals(ConnectType.ARROW_FLIGHT_SQL)) {
@@ -2060,6 +2080,10 @@ public class StmtExecutor {
}
protected void sendTextResultRow(ResultSet resultSet) throws IOException {
+ sendTextResultRow(resultSet, context.getMysqlChannel());
+ }
+
+ protected void sendTextResultRow(ResultSet resultSet, MysqlChannel
channel) throws IOException {
for (List<String> row : resultSet.getResultRows()) {
serializer.reset();
for (String item : row) {
@@ -2069,11 +2093,15 @@ public class StmtExecutor {
serializer.writeLenEncodedString(item);
}
}
- context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
+ channel.sendOnePacket(serializer.toByteBuffer());
}
}
protected void sendBinaryResultRow(ResultSet resultSet) throws IOException
{
+ sendBinaryResultRow(resultSet, context.getMysqlChannel());
+ }
+
+ protected void sendBinaryResultRow(ResultSet resultSet, MysqlChannel
channel) throws IOException {
//
https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_binary_resultset.html#sect_protocol_binary_resultset_row_value
ResultSetMetaData metaData = resultSet.getMetaData();
int nullBitmapLength = (metaData.getColumnCount() + 7 + 2) / 8;
@@ -2136,7 +2164,7 @@ public class StmtExecutor {
}
}
}
- context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
+ channel.sendOnePacket(serializer.toByteBuffer());
}
}
@@ -2311,6 +2339,31 @@ public class StmtExecutor {
planner = new NereidsPlanner(statementContext);
planner.plan(adapter, context.getSessionVariable().toThrift());
+ // A plan that FE can compute on its own (e.g. the empty delta of
a dry run) must be
+ // answered by the frontend, like a regular query does. Otherwise
the coordinator
+ // would send fragments to the placeholder backend registered when
no backend is
+ // needed (see NereidsPlanner#notNeedBackend), which cannot
resolve. Results go to the
+ // caller's channel: the executor's own channel is not connected
to that client.
+ if (context.supportHandleByFe()) {
+ Optional<ResultSet> resultSet =
planner.handleQueryInFe(adapter);
+ if (resultSet.isPresent()) {
+ boolean sendToChannel = !collectMode;
+ if (sendToChannel) {
+ sendResultSet(resultSet.get(),
adapter.getFieldInfos(), sendChannel);
+ }
+ isHandleQueryInFe = true;
+ if (context.getSessionVariable().enableProfile() &&
profile != null) {
+
profile.getSummaryProfile().setExecutedByFrontend(true);
+ }
+ if (sendToChannel) {
+ return new ArrayList<>();
+ }
+ return resultSet.get().getResultRows().stream()
+ .map(ResultRow::new)
+ .collect(Collectors.toList());
+ }
+ }
+
if (!collectMode) {
executeAndSendResult(false, false, adapter, sendChannel, null,
null);
return new ArrayList<>();
diff --git a/regression-test/data/mtmv_p0/ivm/test_ivm_refresh_dry_run.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_refresh_dry_run.out
index 1208b149aa3..686e463d2f3 100644
--- a/regression-test/data/mtmv_p0/ivm/test_ivm_refresh_dry_run.out
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_refresh_dry_run.out
@@ -5,8 +5,8 @@
3 1 30
-- !ivm_dry_run_full --
-1 1 1 15 1 1 4097 0
-4 4 1 40 1 1 4097 0
+1 1 1 15 1 1 [regression-fake-sequence] 0
+4 4 1 40 1 1 [regression-fake-sequence] 0
-- !ivm_dry_run_after --
1 1 10
@@ -14,8 +14,8 @@
3 1 30
-- !ivm_dry_run_repeat --
-1 1 1 15 1 1 4097 0
-4 4 1 40 1 1 4097 0
+1 1 1 15 1 1 [regression-fake-sequence] 0
+4 4 1 40 1 1 [regression-fake-sequence] 0
-- !ivm_dry_run_after_refresh --
1 1 15
@@ -23,3 +23,7 @@
3 1 30
4 1 40
+-- !ivm_dry_run_empty --
+
+-- !ivm_dry_run_empty_repeat --
+
diff --git
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
index d906c610ed8..6ef166d9459 100644
---
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
+++
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
@@ -1727,8 +1727,10 @@ class Suite implements GroovyInterceptable {
return result
}
- // rowConverter: { row -> convertedRow }
+ // rowConverter: { row -> convertedRow }, or { row, meta -> convertedRow }
to inspect the result
+ // metadata, for example to mask a column whose value depends on the
deployment mode.
void quickRunTest(String tag, Object arg, boolean isOrder = false, Closure
rowConverter = null) {
+ boolean converterNeedsMeta = rowConverter != null &&
rowConverter.maximumNumberOfParameters > 1
if (context.config.generateOutputFile ||
context.config.forceGenerateOutputFile) {
Tuple2<List<List<Object>>, ResultSetMetaData> tupleResult = null
if (arg instanceof PreparedStatement) {
@@ -1763,7 +1765,9 @@ class Suite implements GroovyInterceptable {
}
def (result, meta) = tupleResult
if (rowConverter != null) {
- result = result.collect { rowConverter.call(it) }
+ result = result.collect {
+ converterNeedsMeta ? rowConverter.call(it, meta) :
rowConverter.call(it)
+ }
}
if (isOrder) {
result = sortByToString(result)
@@ -1815,7 +1819,9 @@ class Suite implements GroovyInterceptable {
}
def (realResults, meta) = tupleResult
if (rowConverter != null) {
- realResults = realResults.collect { rowConverter.call(it) }
+ realResults = realResults.collect {
+ converterNeedsMeta ? rowConverter.call(it, meta) :
rowConverter.call(it)
+ }
}
if (isOrder) {
realResults = sortByToString(realResults)
diff --git a/regression-test/plugins/plugin_planner.groovy
b/regression-test/plugins/plugin_planner.groovy
index 0a5a12c402a..eca9594bd2d 100644
--- a/regression-test/plugins/plugin_planner.groovy
+++ b/regression-test/plugins/plugin_planner.groovy
@@ -135,3 +135,24 @@ Suite.metaClass.explainIvmPlan = { String tag, String sql
->
)
}
+// __DORIS_SEQUENCE_COL__ of an IVM dry-run delta encodes the refresh version,
whose base value
+// differs between cloud and shared-nothing deployments. Mask that column so
that both modes can
+// share one suite and one .out file.
+def maskIvmDryRunSequence = { row, meta ->
+ for (int i = 1; i <= meta.getColumnCount(); i++) {
+ if ("__DORIS_SEQUENCE_COL__".equalsIgnoreCase(meta.getColumnLabel(i)))
{
+ def masked = new ArrayList(row)
+ masked.set(i - 1, "[regression-fake-sequence]")
+ return masked
+ }
+ }
+ return row
+}
+
+// Named ivm_dry_run_qt instead of order_qt_*: Suite.invokeMethod intercepts
every method whose
+// name starts with qt_ / order_qt_ and derives the tag from the method name,
so a metaClass
+// method could never be reached under those prefixes.
+Suite.metaClass.ivm_dry_run_qt = { String tag, String sql ->
+ delegate.quickRunTest(tag, sql, true, maskIvmDryRunSequence)
+}
+
diff --git a/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy
index 5cb89692a6f..f2583e14e8d 100644
--- a/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy
@@ -16,14 +16,9 @@
// under the License.
suite("test_ivm_refresh_dry_run") {
- // Cloud mode: __DORIS_SEQUENCE_COL__ in the dry-run delta rows derives
from cloud txn
- // versioning and differs from local (e.g. 6145 vs 4097), so the .out
values
- // for the sequence column do not apply.
- if (isCloudMode()) {
- logger.info("skip test_ivm_refresh_dry_run on cloud mode: " +
- "__DORIS_SEQUENCE_COL__ differs between cloud and local")
- return
- }
+ // __DORIS_SEQUENCE_COL__ of the dry-run delta rows encodes the refresh
version, whose base
+ // value differs between cloud and shared-nothing deployments (e.g. 6145
vs 4097).
+ // ivm_dry_run_qt masks that column, so cloud and local share this suite
and its .out.
sql "DROP MATERIALIZED VIEW IF EXISTS test_ivm_refresh_dry_run_mv"
sql "DROP TABLE IF EXISTS test_ivm_refresh_dry_run_base"
@@ -65,9 +60,9 @@ suite("test_ivm_refresh_dry_run") {
order_qt_ivm_dry_run_before "SELECT k1, cnt, sum_v1 FROM
test_ivm_refresh_dry_run_mv"
- order_qt_ivm_dry_run_full """
+ ivm_dry_run_qt("ivm_dry_run_full", """
REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL WITH
DRY RUN
- """
+ """)
// The delta rows picked by a LIMIT depend on scan order (no ORDER BY
before the cap),
// 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") {
order_qt_ivm_dry_run_after "SELECT k1, cnt, sum_v1 FROM
test_ivm_refresh_dry_run_mv"
- order_qt_ivm_dry_run_repeat """
+ ivm_dry_run_qt("ivm_dry_run_repeat", """
REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL WITH
DRY RUN LIMIT 10
- """
+ """)
sql "REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL"
waitingMTMVTaskFinishedByMvName("test_ivm_refresh_dry_run_mv")
order_qt_ivm_dry_run_after_refresh "SELECT k1, cnt, sum_v1 FROM
test_ivm_refresh_dry_run_mv"
+
+ // Empty delta: the delta query is a LogicalEmptyRelation and the dry run
must return an
+ // empty result instead of sending fragments to a placeholder backend.
Every dry-run
+ // comparison goes through ivm_dry_run_qt so the sequence column is masked
in all of them.
+ ivm_dry_run_qt("ivm_dry_run_empty", """
+ REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL WITH
DRY RUN
+ """)
+ ivm_dry_run_qt("ivm_dry_run_empty_repeat", """
+ REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL WITH
DRY RUN
+ """)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
