This is an automated email from the ASF dual-hosted git repository.
jacktengg 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 0f987b6d243 [fix](be) Preserve shared columns in distinct streaming
pass-through (#68880)
0f987b6d243 is described below
commit 0f987b6d24314a28a968449c718a008c8709a0fd
Author: TengJianPing <[email protected]>
AuthorDate: Mon Oct 12 10:28:48 2026 +0800
[fix](be) Preserve shared columns in distinct streaming pass-through
(#68880)
Issue Number: N/A
Related PR: N/A
Problem Summary: EXCEPT DISTINCT can introduce a streaming distinct
aggregation whose grouping expressions repeat an input column, for
example (s, d, d). When preaggregation switches to pass-through, both
output construction and column reuse call assert_mutable() even though
they only transfer column references. Retaining output references can
make a repeated key shared, so a subsequent assertion fails with
COW::assert_mutable: use_count() > 1:
```
W 2026-10-09 06:31:59,596 13803 exception.cpp:34] meet exception, error
code: 6, message: COW::assert_mutable: use_count() > 1
0# doris::Exception::Exception(int, std::basic_string_view<char,
std::char_traits<char> > const&) at
/usr/local/ldb-toolchain-v0.26/bin/../lib/gcc/x86_64-pc-linux-gnu/15/include/g++-v15/bits/basic_string.h:239
1#
doris::DistinctStreamingAggLocalState::_distinct_pre_agg_with_serialized_key(doris::Block*,
doris::Block*) at ../src/core/cow.h:325
2# doris::DistinctStreamingAggOperatorX::push(doris::RuntimeState*,
doris::Block*, bool) const at ../src/common/status.h:519
3#
doris::StatefulOperatorX<doris::DistinctStreamingAggLocalState>::get_block_impl(doris::RuntimeState*,
doris::Block*, bool*) at ../src/common/status.h:519
4# doris::OperatorXBase::get_block(doris::RuntimeState*, doris::Block*,
bool*) at ../src/common/status.h:519
5# doris::OperatorXBase::get_block_after_projects(doris::RuntimeState*,
doris::Block*, bool*) at ./be/build_RELEASE/../src/exec/operator/operator.cpp:0
6# doris::PipelineTask::execute(bool*) at
./be/src/exec/pipeline/pipeline_task.cpp:0
7# doris::TaskScheduler::_do_work(int) at
./be/src/exec/pipeline/task_scheduler.cpp:155
```
Use immutable get_ptr() references in both pass-through paths. The
evaluated key pointers preserve the original column when a repeated
input position has already been replaced during column reuse, and
downstream mutations continue to use copy-on-write. No column data is
copied during pass-through. Add unit coverage with three references to
one input slot for fresh and reused output blocks, transition from
deduplication, retained output ownership, and nullable output. Add
regression coverage for repeated keys in EXCEPT DISTINCT, nullable
EXCEPT output, INTERSECT DISTINCT, and UNION DISTINCT.
Fix queries with repeated projected columns failing with a COW assertion
during distinct streaming aggregation.
根因是：**Distinct Streaming Aggregation 在透传列时调用了要求“列对象独占”的
`assert_mutable()`，而重复分组键会让多个输出位置共享同一个列对象，导致后续调用违反这个要求。**
下面结合修复前的代码，分别说明两条透传路径。
**1\. `assert_mutable()` 检查的是列对象的引用计数**
COW::assert\_mutable() (be/src/core/cow.h:323) 的实现是：
```
MutablePtr assert_mutable() const {
if (this->use_count() > 1) {
throw Exception(ErrorCode::INTERNAL_ERROR,
"COW::assert_mutable: use_count() > 1");
}
return const_cast<COW*>(this)->get_ptr();
}
```
它要求调用时 `use_count() <= 1`，直接返回可变指针，**不会自动复制共享的列对象**。
如果确实需要修改可能共享的列，应使用 IColumn::mutate()
(be/src/core/column/column.h:618)：它通过 `shallow_mutate()` 在引用计数大于 1
时复制列，使修改发生在独立对象上。
这里的引用计数统计的是持有列对象的智能指针数量。例如：
```
in_block.column[1] ─┐
├── 同一个日期列对象 C
output.column[2] ───┘
C.use_count() = 2
```
**2\. 重复投影如何变成重复输入列引用**
例如：
```
SELECT s, d, d FROM t1
EXCEPT DISTINCT
SELECT s, d, d FROM t2;
```
当优化器为集合操作的输入推导局部去重时，InferSetOperatorDistinct
(fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/InferSetOperatorDistinct.java:60)
使用子节点输出构造分组键：
```
new LogicalAggregate<>(
ImmutableList.copyOf(child.getOutput()), true, child)
```
因此，生成的局部聚合可能保留重复槽位，例如：
```
group by: s[#1], d[#2], d[#2]
```
BE 在准备分组列
(be/src/exec/operator/distinct\_streaming\_aggregation\_operator.cpp:177)时，记录表达式执行结果：
```
RETURN_IF_ERROR(_probe_expr_ctxs[i]->execute(in_block, &result_column_id));
// 准备列……
key_columns[i] = in_block->get_by_position(result_column_id).column.get();
result_idxs[i] = result_column_id;
```
假设输入 Block 只有 `s`、`d` 两列，最终得到：
```
in_block: [Cs, Cd]
result_idxs: [ 0, 1, 1]
key_columns: [Cs*, Cd*, Cd*]
```
两个日期键指向同一个列对象。`key_columns` 保存的是裸指针，本身不增加引用计数。
当预聚合因内存限制或去重收益不足设置 `_stop_emplace_flag = true` 后，算子开始透传输入，进入下面的问题路径。
**3\. 首次构造输出：两个重复引用即可报错**
修复前，不复用输出列的透传代码是：
```
for (int i = 0; i < key_size; ++i) {
if (_stop_emplace_flag) {
columns_with_schema.emplace_back(
key_columns[i]->assert_mutable(),
/* type, name */);
}
}
```
对应当前代码的输出构造位置
(be/src/exec/operator/distinct\_streaming\_aggregation\_operator.cpp:267)。
以 `[s, d, d]` 为例，假设日期列 `Cd` 开始只有输入 Block 持有：
| 时刻 | 持有 `Cd` 的位置 | 稳定引用计数 |
|---|---|---:|
| 处理第一个 `d` 前 | 输入列 | 1 |
| 第一个 `d` 加入输出后 | 输入列、第一个日期输出列 | 2 |
| 处理第二个 `d` 时 | 再次调用 `Cd->assert_mutable()` | 2，抛异常 |
第一个 `d` 输出时增加了一个列引用，输入引用仍然存在。第二个 `d` 再要求独占，就会失败。
后面的 `in_block->clear()` 原本可以释放输入引用，但异常已经在循环内发生，执行不到那里。
**4\. 复用输出 Block：三个重复引用能展示同样的问题**
修复前，复用输出列的透传位置
(be/src/exec/operator/distinct\_streaming\_aggregation\_operator.cpp:235)
是：
```
for (int i = 0; i < key_size; ++i) {
auto output_column = out_block->get_by_position(i).column;
out_block->replace_by_position(i, key_columns[i]->assert_mutable());
in_block->replace_by_position(result_idxs[i], output_column);
}
```
这个分支将旧输出列放回输入 Block，释放输入对原数据列的引用。
以 `[s, d, d, d]` 为例：
```
result_idxs = [0, 1, 1, 1]
```
假设输出列最初为空，且没有其他所有者：
| 时刻 | 原日期列 `Cd` 的所有者 | 稳定引用计数 |
|---|---|---:|
| 第一个 `d` 处理结束 | 第一个日期输出列；输入位置已换成旧输出空列 | 1 |
| 第二个 `d` 处理结束 | 第一个、第二个日期输出列 | 2 |
| 第三个 `d` 开始处理 | 调用 `Cd->assert_mutable()` | 2，抛异常 |
第二次替换的输入位置已经是空列，无法再释放原日期列的引用。于是两个输出列共同持有 `Cd`，第三次独占断言失败。
因此，**复用分支中只有两个相同槽位不一定立即报错**；已有单测使用 `[0, 1, 0, 0]`，让同一输入槽位出现三次，覆盖这一过程。
**5\. 为什么改为 `get_ptr()` 能解决**
两条透传路径现在都使用：
```
key_columns[i]->get_ptr()
```
`key_columns` 的类型是 `std::vector<const IColumn*>`，因此这里调用的是 const 版本的
`get_ptr()` (be/src/core/cow.h:304)，返回不可变列指针。
透传只需要让输出引用已有列数据，允许多个输出位置共享该列。后续需要修改数据时，仍通过 `IColumn::mutate()` 执行 COW。
复用分支继续从预先记录的 `key_columns` 取指针也很关键：第一次替换后，`in_block[result_idxs[i]]`
已经变成旧输出列，而 `key_columns[i]` 仍指向原数据列，并由前面的输出位置保持其生命周期。
关于 GROUP BY，直接的 `GROUP BY s, d, d` 会在 NormalizeAggregate
(fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/NormalizeAggregate.java:167)
中去重。静态分析发现，`GROUP BY + LIMIT` 下推到 `UNION ALL` 分支时，槽位映射可能重新引入重复键；已在回归用例
(regression-test/suites/query\_p0/set\_operations/distinct\_streaming\_repeated\_keys.groovy:101)中补充直接分组对照和该候选场景，并检查计划中同一槽位是否出现三次。提交为
`c8c1f67d3e9`，按你的要求未执行新增用例，因此尚未实测确认该 GROUP BY 场景。
- Test: Unit Test / Regression test / Manual test
- ASAN build with ./build.sh --be -j 12
- 11 BE unit tests covering distinct streaming aggregation
- Four regression queries; results also match non-streaming execution
- Original reproducing SQL on the supplied table data
- clang-format 16, format check, build hygiene, and clang-tidy on
changed lines
- Behavior changed: Yes; repeated grouping keys pass through without an
incorrect ownership assertion
- Does this need documentation: No
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../distinct_streaming_aggregation_operator.cpp | 7 +-
...istinct_streaming_aggregation_operator_test.cpp | 68 ++++++++++
.../distinct_streaming_repeated_keys.out | 13 ++
.../distinct_streaming_repeated_keys.groovy | 147 +++++++++++++++++++++
4 files changed, 232 insertions(+), 3 deletions(-)
diff --git a/be/src/exec/operator/distinct_streaming_aggregation_operator.cpp
b/be/src/exec/operator/distinct_streaming_aggregation_operator.cpp
index c9b61f1f832..45ab5827fb7 100644
--- a/be/src/exec/operator/distinct_streaming_aggregation_operator.cpp
+++ b/be/src/exec/operator/distinct_streaming_aggregation_operator.cpp
@@ -230,10 +230,11 @@ Status
DistinctStreamingAggLocalState::_distinct_pre_agg_with_serialized_key(
DCHECK_EQ(out_block->columns(), key_size);
if (_stop_emplace_flag && _distinct_row.empty()) {
// If _stop_emplace_flag is true and _distinct_row is also empty,
it means it is in streaming mode, outputting what is input
- // swap the column directly, to solve Check failed:
d.column->use_count() == 1 (2 vs. 1)
+ // Share the evaluated keys as immutable columns. A repeated input
position may
+ // already have been replaced, but its column is retained by the
earlier output key.
for (int i = 0; i < key_size; ++i) {
auto output_column = out_block->get_by_position(i).column;
- out_block->replace_by_position(i,
key_columns[i]->assert_mutable());
+ out_block->replace_by_position(i, key_columns[i]->get_ptr());
in_block->replace_by_position(result_idxs[i], output_column);
}
} else {
@@ -266,7 +267,7 @@ Status
DistinctStreamingAggLocalState::_distinct_pre_agg_with_serialized_key(
ColumnsWithTypeAndName columns_with_schema;
for (int i = 0; i < key_size; ++i) {
if (_stop_emplace_flag) {
-
columns_with_schema.emplace_back(key_columns[i]->assert_mutable(),
+ columns_with_schema.emplace_back(key_columns[i]->get_ptr(),
_probe_expr_ctxs[i]->root()->data_type(),
_probe_expr_ctxs[i]->root()->expr_name());
} else {
diff --git
a/be/test/exec/operator/distinct_streaming_aggregation_operator_test.cpp
b/be/test/exec/operator/distinct_streaming_aggregation_operator_test.cpp
index 7c3e732925f..f52a65ba186 100644
--- a/be/test/exec/operator/distinct_streaming_aggregation_operator_test.cpp
+++ b/be/test/exec/operator/distinct_streaming_aggregation_operator_test.cpp
@@ -295,4 +295,72 @@ TEST_F(DistinctStreamingAggOperatorTest,
pass_through_does_not_consume_pushed_li
EXPECT_EQ(local_state->_aggregated_block->rows(), 4);
}
+struct DistinctStreamingAggRepeatedKeyTest : public
DistinctStreamingAggOperatorTest,
+ public
::testing::WithParamInterface<bool> {
+ void SetUp() override {
+ DistinctStreamingAggOperatorTest::SetUp();
+ op->_is_streaming_preagg = true;
+ auto input_type = std::make_shared<DataTypeInt64>();
+ auto output_type = GetParam() ? make_nullable(input_type) : input_type;
+ create_op({input_type, input_type, input_type, input_type},
+ {output_type, output_type, output_type, output_type});
+ local_state->_probe_expr_ctxs = {MockSlotRef::create_mock_context(0,
input_type),
+ MockSlotRef::create_mock_context(1,
input_type),
+ MockSlotRef::create_mock_context(0,
input_type),
+ MockSlotRef::create_mock_context(0,
input_type)};
+ }
+
+ Block expected_block(const std::vector<Int64>& first, const
std::vector<Int64>& second) {
+ auto block = ColumnHelper::create_block<DataTypeInt64>(first, second);
+ block.insert(block.get_by_position(0));
+ block.insert(block.get_by_position(0));
+ if (GetParam()) {
+ for (size_t i = 0; i < block.columns(); ++i) {
+ auto& column = block.get_by_position(i);
+ column.column = make_nullable(column.column);
+ column.type = make_nullable(column.type);
+ }
+ }
+ return block;
+ }
+};
+
+TEST_P(DistinctStreamingAggRepeatedKeyTest, pass_through) {
+ local_state->_stop_emplace_flag = true;
+ auto input = ColumnHelper::create_block<DataTypeInt64>({1, 1, 2}, {10, 10,
20});
+ ASSERT_TRUE(op->push(state.get(), &input, false));
+
+ auto output = expected_block({}, {});
+ bool eos = false;
+ ASSERT_TRUE(op->pull(state.get(), &output, &eos));
+ EXPECT_TRUE(ColumnHelper::block_equal(output, expected_block({1, 1, 2},
{10, 10, 20})));
+
+ // The output schema is now reused. Keep the previous output alive to
check its ownership.
+ input = ColumnHelper::create_block<DataTypeInt64>({3, 4, 4}, {30, 40, 40});
+ ASSERT_TRUE(op->push(state.get(), &input, false));
+ auto next_output = expected_block({}, {});
+ ASSERT_TRUE(op->pull(state.get(), &next_output, &eos));
+ EXPECT_TRUE(ColumnHelper::block_equal(next_output, expected_block({3, 4,
4}, {30, 40, 40})));
+ EXPECT_TRUE(ColumnHelper::block_equal(output, expected_block({1, 1, 2},
{10, 10, 20})));
+}
+
+TEST_P(DistinctStreamingAggRepeatedKeyTest, pass_through_after_deduplication) {
+ auto input = ColumnHelper::create_block<DataTypeInt64>({1, 1, 2}, {10, 10,
20});
+ ASSERT_TRUE(op->push(state.get(), &input, false));
+ EXPECT_EQ(local_state->_aggregated_block->rows(), 2);
+ auto deduplicated_output = expected_block({}, {});
+ bool eos = false;
+ ASSERT_TRUE(op->pull(state.get(), &deduplicated_output, &eos));
+ EXPECT_TRUE(ColumnHelper::block_equal(deduplicated_output,
expected_block({1, 2}, {10, 20})));
+
+ local_state->_stop_emplace_flag = true;
+ input = ColumnHelper::create_block<DataTypeInt64>({3, 3, 4}, {30, 30, 40});
+ ASSERT_TRUE(op->push(state.get(), &input, false));
+ auto output = expected_block({}, {});
+ ASSERT_TRUE(op->pull(state.get(), &output, &eos));
+ EXPECT_TRUE(ColumnHelper::block_equal(output, expected_block({3, 3, 4},
{30, 30, 40})));
+}
+
+INSTANTIATE_TEST_SUITE_P(NullableOutput, DistinctStreamingAggRepeatedKeyTest,
::testing::Bool());
+
} // namespace doris
diff --git
a/regression-test/data/query_p0/set_operations/distinct_streaming_repeated_keys.out
b/regression-test/data/query_p0/set_operations/distinct_streaming_repeated_keys.out
new file mode 100644
index 00000000000..c61697af12f
--- /dev/null
+++
b/regression-test/data/query_p0/set_operations/distinct_streaming_repeated_keys.out
@@ -0,0 +1,13 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !except --
+1638 2026-10-01 2026-10-01
+
+-- !except_nullable --
+2457 2457 2457 2026-10-01 2026-10-01
+
+-- !intersect --
+1639 2026-10-01 2026-10-01
+
+-- !union --
+4915 2026-10-01 2026-10-01
+
diff --git
a/regression-test/suites/query_p0/set_operations/distinct_streaming_repeated_keys.groovy
b/regression-test/suites/query_p0/set_operations/distinct_streaming_repeated_keys.groovy
new file mode 100644
index 00000000000..f7c57617733
--- /dev/null
+++
b/regression-test/suites/query_p0/set_operations/distinct_streaming_repeated_keys.groovy
@@ -0,0 +1,147 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("distinct_streaming_repeated_keys", "query,p0") {
+ sql "DROP TABLE IF EXISTS distinct_streaming_repeated_keys"
+ sql """
+ CREATE TABLE distinct_streaming_repeated_keys (
+ id INT NOT NULL,
+ s STRING NULL,
+ d DATE NOT NULL
+ ) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+ sql """
+ INSERT INTO distinct_streaming_repeated_keys
+ SELECT number, IF(number % 5 = 0, NULL, CAST(number AS STRING)), DATE
'2026-10-01'
+ FROM numbers("number" = "8192")
+ """
+
+ // Favor local distinct under the set operation, with a shuffle because id
is not a key.
+ sql """
+ ALTER TABLE distinct_streaming_repeated_keys MODIFY COLUMN s
+ SET STATS ('row_count' = '8192', 'ndv' = '1', 'num_nulls' = '1639',
+ 'min_value' = '0', 'max_value' = '8191')
+ """
+ sql """
+ ALTER TABLE distinct_streaming_repeated_keys MODIFY COLUMN d
+ SET STATS ('row_count' = '8192', 'ndv' = '1', 'num_nulls' = '0',
+ 'min_value' = '2026-10-01', 'max_value' = '2026-10-01')
+ """
+ sql "SET agg_phase = 2"
+ sql "SET disable_streaming_preaggregations = false"
+ sql "SET enable_distinct_streaming_aggregation = true"
+ sql "SET parallel_pipeline_task_num = 1"
+ sql "SET batch_size = 256"
+ sql "SET enable_spill = true"
+ // Switch to pass-through after the first batch has populated the hash
table.
+ sql "SET spill_streaming_agg_mem_limit = 1"
+
+ explain {
+ sql """
+ SELECT s, d, d FROM distinct_streaming_repeated_keys WHERE id <
4096
+ EXCEPT DISTINCT
+ SELECT s, d, d FROM distinct_streaming_repeated_keys WHERE id % 2
= 0
+ """
+ contains "STREAMING"
+ }
+
+ order_qt_except """
+ SELECT COUNT(*), MIN(d1), MAX(d2) FROM (
+ SELECT s, d AS d1, d AS d2
+ FROM distinct_streaming_repeated_keys WHERE id < 4096
+ EXCEPT DISTINCT
+ SELECT s, d, d FROM distinct_streaming_repeated_keys WHERE id % 2
= 0
+ ) t
+ """
+
+ // The first date key becomes nullable, exercising the output path without
memory reuse.
+ order_qt_except_nullable """
+ SELECT COUNT(*), COUNT(d1), COUNT(d2), MIN(d1), MAX(d2) FROM (
+ SELECT s, d AS d1, d AS d2
+ FROM distinct_streaming_repeated_keys WHERE id < 4096
+ EXCEPT DISTINCT
+ SELECT s, IF(id % 4 = 0, NULL, d), d
+ FROM distinct_streaming_repeated_keys WHERE id % 2 = 0
+ ) t
+ """
+
+ order_qt_intersect """
+ SELECT COUNT(*), MIN(d1), MAX(d2) FROM (
+ SELECT s, d AS d1, d AS d2
+ FROM distinct_streaming_repeated_keys WHERE id < 4096
+ INTERSECT DISTINCT
+ SELECT s, d, d FROM distinct_streaming_repeated_keys WHERE id % 2
= 0
+ ) t
+ """
+
+ order_qt_union """
+ SELECT COUNT(*), MIN(d1), MAX(d2) FROM (
+ SELECT s, d AS d1, d AS d2
+ FROM distinct_streaming_repeated_keys WHERE id < 4096
+ UNION DISTINCT
+ SELECT s, d, d FROM distinct_streaming_repeated_keys WHERE id % 2
= 0
+ ) t
+ """
+
+ // NormalizeAggregate removes repeated keys in a direct GROUP BY. Keep
this as
+ // a control case, rather than assuming that repeated SQL text repeats
input slots.
+ check_sqls_result_equal("""
+ SELECT s, d, d, d FROM distinct_streaming_repeated_keys
+ GROUP BY s, d, d, d
+ ORDER BY 1, 2, 3, 4
+ """, """
+ SELECT s, d, d, d FROM distinct_streaming_repeated_keys
+ GROUP BY s, d
+ ORDER BY 1, 2, 3, 4
+ """)
+
+ // PushDownLimitDistinctThroughUnion maps distinct union output slots back
to
+ // the same branch slot after aggregate normalization. Three references to
d
+ // exercise shared ownership even when the output Block is reused.
+ // Use ordinals to group all four union columns despite their repeated
names.
+ // The limit covers every input row, so it cannot make the result
nondeterministic.
+ def groupByUnionAllLimit = """
+ SELECT * FROM (
+ SELECT * FROM (
+ SELECT s, d, d, d FROM distinct_streaming_repeated_keys WHERE
id < 4096
+ UNION ALL
+ SELECT s, d, d, d FROM distinct_streaming_repeated_keys WHERE
id % 2 = 0
+ ) u
+ GROUP BY 1, 2, 3, 4
+ LIMIT 8192
+ ) grouped
+ ORDER BY 1, 2, 3, 4
+ """
+
+ explain {
+ sql groupByUnionAllLimit
+ verbose true
+ check { plan ->
+ // Require three references to one input slot in a streaming
aggregate,
+ // not merely different slots projecting identical column values.
+ (plan =~ /STREAMING\s*\|\s*group by: s\[#\d+\], d\[#(\d+)\],
d\[#\1\], d\[#\1\]/).find()
+ }
+ }
+
+ check_sqls_result_equal(groupByUnionAllLimit, """
+ SELECT s, d, d, d FROM distinct_streaming_repeated_keys
+ WHERE id < 4096 OR id % 2 = 0
+ GROUP BY s, d
+ ORDER BY 1, 2, 3, 4
+ """)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
