Merge pull request ClickHouse#76094 from ClickHouse/backport/24.8/75512

Backport ClickHouse#75512 to 24.8: Fix check for nested elements arrays sizes in case of distributed JOIN and JSON subcolumns

Author: Avogar

src/Interpreters/InterpreterInsertQuery.cpp

@@ -33,6 +33,7 @@
 #include <Processors/Transforms/SquashingTransform.h>
 #include <Processors/Transforms/PlanSquashingTransform.h>
 #include <Processors/Transforms/getSourceFromASTInsertQuery.h>
+#include <Processors/Transforms/NestedElementsValidationTransform.h>
 #include <Processors/QueryPlan/QueryPlan.h>
 #include <QueryPipeline/QueryPipelineBuilder.h>
 #include <Storages/StorageDistributed.h>
@@ -380,6 +381,12 @@ Chain InterpreterInsertQuery::buildPreSinkChain(
 
     /// Note that we wrap transforms one on top of another, so we write them in reverse of data processing order.
 
+    /// Add transform to check if the sizes of arrays - elements of nested data structures doesn't match.
+    /// We have to make this assertion before writing to table, because storage engine may assume that they have equal sizes.
+    /// NOTE It'd better to do this check in serialization of nested structures (in place when this assumption is required),
+    /// but currently we don't have methods for serialization of nested structures "as a whole".
+    out.addSource(std::make_shared<NestedElementsValidationTransform>(input_header()));
+
     /// Checking constraints. It must be done after calculation of all defaults, so we can check them on calculated columns.
     if (const auto & constraints = metadata_snapshot->getConstraints(); !constraints.empty())
         out.addSource(std::make_shared<CheckConstraintsTransform>(

src/Interpreters/inplaceBlockConversions.cpp

@@ -257,18 +257,6 @@ static std::unordered_map<String, ColumnPtr> collectOffsetsColumns(
 
                     if (offsets_column->size() != current_offsets_column->size() && inside_variant_element)
                         offsets_column = offsets_column->size() < current_offsets_column->size() ? offsets_column : current_offsets_column;
-#ifndef NDEBUG
-                    else
-                    {
-                        const auto & offsets_data = assert_cast<const ColumnUInt64 &>(*offsets_column).getData();
-                        const auto & current_offsets_data = assert_cast<const ColumnUInt64 &>(*current_offsets_column).getData();
-
-                        if (offsets_data != current_offsets_data)
-                            throw Exception(ErrorCodes::LOGICAL_ERROR,
-                                            "Found non-equal columns with offsets (sizes: {} and {}) for stream {}",
-                                            offsets_data.size(), current_offsets_data.size(), stream_name);
-                    }
-#endif
                 }
             }
         }, available_column->type, res_columns[i]);

src/Processors/Sinks/SinkToStorage.cpp

@@ -1,5 +1,4 @@
 #include <Processors/Sinks/SinkToStorage.h>
-#include <DataTypes/NestedUtils.h>
 
 namespace DB
 {
@@ -8,13 +7,6 @@ SinkToStorage::SinkToStorage(const Block & header) : ExceptionKeepingTransform(h
 
 void SinkToStorage::onConsume(Chunk chunk)
 {
-    /** Throw an exception if the sizes of arrays - elements of nested data structures doesn't match.
-      * We have to make this assertion before writing to table, because storage engine may assume that they have equal sizes.
-      * NOTE It'd better to do this check in serialization of nested structures (in place when this assumption is required),
-      * but currently we don't have methods for serialization of nested structures "as a whole".
-      */
-    Nested::validateArraySizes(getHeader().cloneWithColumns(chunk.getColumns()));
-
     consume(chunk);
     cur_chunk = std::move(chunk);
 }

src/Processors/Transforms/NestedElementsValidationTransform.cpp

@@ -0,0 +1,16 @@
+#include <Processors/Transforms/NestedElementsValidationTransform.h>
+#include <DataTypes/NestedUtils.h>
+
+namespace DB
+{
+
+NestedElementsValidationTransform::NestedElementsValidationTransform(const Block & header) : ISimpleTransform(header, header, false)
+{
+}
+
+void NestedElementsValidationTransform::transform(Chunk & chunk)
+{
+    Nested::validateArraySizes(getOutputPort().getHeader().cloneWithColumns(chunk.getColumns()));
+}
+
+}

src/Processors/Transforms/NestedElementsValidationTransform.h

@@ -0,0 +1,20 @@
+#pragma once
+
+#include <Processors/ISimpleTransform.h>
+
+namespace DB
+{
+
+class NestedElementsValidationTransform : public ISimpleTransform
+{
+public:
+    explicit NestedElementsValidationTransform(const Block & header);
+
+    String getName() const override { return "NestedElementsValidationTransform"; }
+
+protected:
+    void transform(Chunk & chunk) override;
+};
+
+}
+

tests/queries/0_stateless/03325_distributed_join_json_array_subcolumns.reference

@@ -0,0 +1 @@
+0

tests/queries/0_stateless/03325_distributed_join_json_array_subcolumns.sql

@@ -0,0 +1,31 @@
+SET allow_experimental_json_type=1;
+SET allow_experimental_analyzer=1;
+
+DROP TABLE IF EXISTS test_distr;
+DROP TABLE IF EXISTS test;
+
+CREATE TABLE test
+(
+    id Int64,
+    data JSON(arr1 Array(String), arr2 Array(Int32))
+)
+ENGINE = MergeTree ORDER BY id;
+
+
+CREATE TABLE test_distr 
+(
+    id Int64,
+    data JSON(arr1 Array(String), arr2 Array(Int32))
+)
+ENGINE = Distributed(test_cluster_one_shard_three_replicas_localhost, currentDatabase(), 'test', murmurHash2_32(id));
+
+INSERT INTO test FORMAT Values (1, '{"arr1" : ["s1", "s2", "s3"], "arr2" : []}'), (2, '{"arr1" : ["s4", "s5"], "arr2" : [42]}');
+
+SELECT count()
+FROM test_distr as left
+GLOBAL INNER JOIN test_distr as right on left.id = right.id
+WHERE has(right.data.arr1, 's3') AND has(right.data.arr2, 42);
+
+DROP TABLE test_distr;
+DROP TABLE test;
+