Merge branch 'master' into remove_settings_join_concurrent_build

dbms/src/Columns/ColumnUtils.cpp

@@ -0,0 +1,33 @@
+// Copyright 2022 PingCAP, Ltd.
+//
+// Licensed 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.
+
+#include <Columns/ColumnUtils.h>
+
+namespace DB
+{
+bool columnEqual(const ColumnPtr & expected, const ColumnPtr & actual, String & unequal_msg)
+{
+    for (size_t i = 0, size = expected->size(); i < size; ++i)
+    {
+        auto expected_field = (*expected)[i];
+        auto actual_field = (*actual)[i];
+        if (expected_field != actual_field)
+        {
+            unequal_msg = fmt::format("Value {} mismatch {} vs {} ", i, expected_field.toString(), actual_field.toString());
+            return false;
+        }
+    }
+    return true;
+}
+} // namespace DB

dbms/src/Columns/ColumnUtils.h

@@ -0,0 +1,22 @@
+// Copyright 2022 PingCAP, Ltd.
+//
+// Licensed 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.
+
+#pragma once
+
+#include <Columns/IColumn.h>
+
+namespace DB
+{
+bool columnEqual(const ColumnPtr & expected, const ColumnPtr & actual, String & unequal_msg);
+} // namespace DB

dbms/src/Core/BlockUtils.cpp

@@ -0,0 +1,65 @@
+// Copyright 2022 PingCAP, Ltd.
+//
+// Licensed 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.
+
+#include <Core/BlockUtils.h>
+
+namespace DB
+{
+bool blockEqual(const Block & expected, const Block & actual, String & unequal_msg)
+{
+    if (size_t rows_a = expected.rows(), rows_b = actual.rows(); rows_a != rows_b)
+    {
+        unequal_msg = fmt::format("Row counter are not equal: {} vs {} ", rows_a, rows_b);
+        return false;
+    }
+
+    size_t size_a = expected.columns();
+    size_t size_b = actual.columns();
+    if (size_a != size_b)
+    {
+        unequal_msg = fmt::format("Columns size are not equal: {} vs {} ", size_a, size_b);
+        return false;
+    }
+
+    for (size_t i = 0; i < size_a; ++i)
+    {
+        bool equal = columnEqual(expected.getByPosition(i).column, actual.getByPosition(i).column, unequal_msg);
+        if (!equal)
+        {
+            unequal_msg = fmt::format("{}th columns are not equal, details: {}", i, unequal_msg);
+            return false;
+        }
+    }
+    return true;
+}
+
+String formatBlockData(const Block & block)
+{
+    size_t rows = block.rows();
+    FmtBuffer buf;
+    for (size_t i = 0; i < rows; ++i)
+    {
+        buf.joinStr(
+            block.begin(),
+            block.end(),
+            [i](const auto & block, FmtBuffer & fb) {
+                auto column = block.column;
+                auto field = (*column)[i];
+                fb.append(field.toString());
+            },
+            ", ");
+    }
+    return buf.toString();
+}
+} // namespace DB

dbms/src/Core/BlockUtils.h

@@ -0,0 +1,25 @@
+// Copyright 2022 PingCAP, Ltd.
+//
+// Licensed 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.
+
+#pragma once
+
+#include <Columns/ColumnUtils.h>
+#include <Core/Block.h>
+
+namespace DB
+{
+bool blockEqual(const Block & expected, const Block & actual, String & unequal_msg);
+
+String formatBlockData(const Block & block);
+} // namespace DB

dbms/src/DataStreams/UniqRawResReformatBlockOutputStream.h

@@ -0,0 +1,75 @@
+// Copyright 2022 PingCAP, Ltd.
+//
+// Licensed 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.
+
+#pragma once
+
+#include <AggregateFunctions/AggregateFunctionUniq.h>
+#include <DataStreams/IProfilingBlockInputStream.h>
+
+namespace DB
+{
+class UniqRawResReformatBlockOutputStream : public IProfilingBlockInputStream
+{
+public:
+    explicit UniqRawResReformatBlockOutputStream(const BlockInputStreamPtr & in_)
+        : in(in_)
+    {}
+
+    String getName() const override { return "UniqRawResReformat"; }
+
+    Block getHeader() const override { return in->getHeader(); }
+
+protected:
+    Block readImpl() override
+    {
+        while (true)
+        {
+            Block block = in->read();
+            if (!block)
+                return block;
+
+            size_t num_columns = block.columns();
+            MutableColumns columns(num_columns);
+            for (size_t i = 0; i < num_columns; ++i)
+            {
+                ColumnWithTypeAndName & ori_column = block.getByPosition(i);
+
+                if (std::string::npos != ori_column.name.find_first_of(uniq_raw_res_name))
+                {
+                    MutableColumnPtr mutable_holder = ori_column.column->cloneEmpty();
+
+                    for (size_t j = 0; j < ori_column.column->size(); ++j)
+                    {
+                        Field field;
+                        ori_column.column->get(j, field);
+
+                        auto & str_ref = field.safeGet<String>();
+
+                        ReadBufferFromString in(str_ref);
+                        AggregateFunctionUniqUniquesHashSetDataForVariadicRawRes set;
+                        set.set.read(in);
+
+                        mutable_holder->insert(std::to_string(set.set.size()));
+                    }
+                    ori_column.column = std::move(mutable_holder);
+                }
+            }
+            return block;
+        }
+    }
+
+private:
+    BlockInputStreamPtr in;
+};
+} // namespace DB