Merge pull request ClickHouse#92253 from ClickHouse/backport/25.8/91886

Backport ClickHouse#91886 to 25.8: Add repair mechanism for JSON with duplicated path in shared data and dynamic paths

Author: Avogar

src/Columns/ColumnObject.cpp

@@ -8,6 +8,7 @@
 #include <IO/ReadBufferFromString.h>
 #include <Common/Arena.h>
 #include <Common/SipHash.h>
+#include <Common/logger_useful.h>
 
 namespace DB
 {
@@ -728,10 +729,8 @@ void ColumnObject::insertFromSharedDataAndFillRemainingDynamicPaths(const DB::Co
             {
                 /// Deserialize binary value into dynamic column from shared data.
                 if (it->second->size() != current_size)
-                {
-                    src_object_column.validateDynamicPathsAndSharedData();
                     throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected size of dynamic path {}: {} != {}", path, it->second->size(), current_size);
-                }
+
                 deserializeValueFromSharedData(src_shared_data_values, i, *it->second);
             }
             else if (auto * dynamic_path_column = tryToAddNewDynamicPath(path))
@@ -2070,28 +2069,121 @@ int ColumnObject::doCompareAt(size_t n, size_t m, const IColumn & rhs, int nan_d
     return 1;
 }
 
-void ColumnObject::validateDynamicPathsAndSharedData(size_t shared_data_offset) const
+void ColumnObject::repairDuplicatesInDynamicPathsAndSharedData(size_t offset)
 {
     if (dynamic_paths.empty())
         return;
 
+    /// First, check if all dynamic paths have correct sizes, just in case.
     size_t expected_size = shared_data->size();
     for (const auto & [path, column] : dynamic_paths)
     {
         if (column->size() != expected_size)
             throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected size of dynamic path {}: {} != {}", path, column->size(), expected_size);
     }
 
+    /// Second, iterate over paths in shared data and check if we have any path that is also present in dynamic paths.
     const auto & shared_data_offsets = getSharedDataOffsets();
-    const auto [shared_data_paths, _] = getSharedDataPathsAndValues();
-    size_t shared_data_paths_start = shared_data_offsets[ssize_t(shared_data_offset) - 1];
-    size_t shared_data_paths_end = shared_data_offsets.back();
-    for (size_t i = shared_data_paths_start; i != shared_data_paths_end; ++i)
+    const auto [shared_data_paths, shared_data_values] = getSharedDataPathsAndValues();
+    /// Remember the first row with duplicates if any. We will start repair from this row.
+    std::optional<size_t> first_row_with_duplicates = std::nullopt;
+    size_t size = shared_data_offsets.size();
+    for (size_t i = offset; i < size; ++i)
+    {
+        size_t shared_data_start = shared_data_offsets[i - 1];
+        size_t shared_data_end = shared_data_offsets[i];
+        for (size_t j = shared_data_start; j < shared_data_end; ++j)
+        {
+            if (dynamic_paths.contains(shared_data_paths->getDataAt(j)))
+            {
+                /// Duplicate is found, no need to iterate further, we need to start repair.
+                first_row_with_duplicates = i;
+                break;
+            }
+        }
+
+        if (first_row_with_duplicates)
+            break;
+    }
+
+    if (!first_row_with_duplicates)
+        return;
+
+    LOG_TRACE(getLogger("ColumnObject"), "Repair duplicates in Object column starting from row {}", *first_row_with_duplicates);
+
+    /// During repair we create new shared data without duplicated dynamic paths
+    /// update corresponding dynamic paths with values from shared data.
+    auto new_shared_data = shared_data->cloneResized(*first_row_with_duplicates);
+    const auto [new_shared_data_paths, new_shared_data_values, new_shared_data_offsets] = getSharedDataPathsValuesAndOffsets(*new_shared_data);
+    new_shared_data_offsets->reserve(size);
+    PathToColumnMap new_dynamic_paths;
+    for (size_t i = *first_row_with_duplicates; i < size; ++i)
     {
-        auto path = shared_data_paths->getDataAt(i);
-        if (dynamic_paths.contains(path))
-            throw Exception(ErrorCodes::LOGICAL_ERROR, "Path {} is present both in dynamic paths and in shared data", path.toString());
+        size_t shared_data_start = shared_data_offsets[i - 1];
+        size_t shared_data_end = shared_data_offsets[i];
+        for (size_t j = shared_data_start; j < shared_data_end; ++j)
+        {
+            auto path = shared_data_paths->getDataAt(j).toView();
+            auto it = dynamic_paths.find(path);
+            if (it == dynamic_paths.end())
+            {
+                new_shared_data_paths->insertFrom(*shared_data_paths, j);
+                new_shared_data_values->insertFrom(*shared_data_values, j);
+            }
+            /// We update dynamic path with value from shared data only if dynamic path has NULL at this row.
+            else if (it->second->isNullAt(i))
+            {
+                auto new_it = new_dynamic_paths.find(path);
+                if (new_it == new_dynamic_paths.end())
+                {
+                    new_it = new_dynamic_paths.emplace(path, it->second->cloneResized(i)).first;
+                    new_it->second->reserve(size);
+                }
+
+                deserializeValueFromSharedData(shared_data_values, j, *new_it->second);
+            }
+            /// Situation when both values in dynamic path and shared data are non-NULL should not be possible
+            /// and we cannot repair it anyhow. Throw logical error exception in this case.
+            else
+            {
+                auto value = shared_data_values->getDataAt(j).toView();
+                ReadBufferFromMemory buf(value.data(), value.size());
+                auto type_from_shared_data = decodeDataType(buf);
+                if (!isNothing(type_from_shared_data))
+                {
+                    auto type_from_dynamic_path = dynamic_paths_ptrs.find(path)->second->getTypeAt(i);
+                    throw Exception(
+                        ErrorCodes::LOGICAL_ERROR,
+                        "Path {} is present both in dynamic paths and shared data and has two non-null values at the row {}."
+                        "Value type in dynamic paths: {}. Value type in shared data: {}",
+                        path,
+                        i,
+                        type_from_dynamic_path->getName(),
+                        type_from_shared_data->getName());
+                }
+            }
+        }
+
+        new_shared_data_offsets->push_back(new_shared_data_paths->size());
+        /// Update new dynamic paths that were not updated with value from shared data in this row.
+        for (auto & [path, column] : new_dynamic_paths)
+        {
+            if (column->size() == i)
+                column->insertFrom(*dynamic_paths.at(path), i);
+        }
     }
+
+    for (auto & [path, column] : new_dynamic_paths)
+    {
+        if (column->size() != size)
+            throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected size of new dynamic path {}: {} != {}", path, column->size(), size);
+        dynamic_paths_ptrs[path] = assert_cast<ColumnDynamic *>(column.get());
+        dynamic_paths[path] = std::move(column);
+    }
+
+    if (new_shared_data->size() != size)
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected size of new shared data: {} != {}", new_shared_data->size(), size);
+    shared_data = std::move(new_shared_data);
 }
 
 }

src/Columns/ColumnObject.h

@@ -282,8 +282,12 @@ class ColumnObject final : public COWHelper<IColumnHelper<ColumnObject>, ColumnO
     /// Insert all the data from shared data with specified path to dynamic column.
     static void fillPathColumnFromSharedData(IColumn & path_column, StringRef path, const ColumnPtr & shared_data_column, size_t start, size_t end);
 
-    /// Validate that all dynamic paths have correct sizes and that shared data doesn't contain any dynamic paths.
-    void validateDynamicPathsAndSharedData(size_t shared_data_offset = 0) const;
+    /// Due to previous bugs we can have an invalid state where we have some path
+    /// both in shared data and in dynamic paths and only one value is not NULL.
+    /// This methods repairs the column and removes this duplicate by removing path
+    /// and value from shared data and keeping only dynamic path containing non-Null value.
+    /// offset argument - is the offset from which we should check for duplicates.
+    void repairDuplicatesInDynamicPathsAndSharedData(size_t offset = 0);
 
 private:
     class SortedPathsIterator;

src/Columns/tests/gtest_column_object.cpp

@@ -414,3 +414,33 @@ TEST(ColumnObject, rollback)
     ASSERT_EQ((*typed_paths.at("a.a"))[0], Field{1u});
     ASSERT_EQ((*dynamic_paths.at("a.c"))[1], Field{"ccc"});
 }
+
+TEST(ColumnObject, RepairDuplicatesInDynamicPathsAndSharedData)
+{
+    auto type_with_dynamic_paths = DataTypeFactory::instance().get("JSON");
+    auto column_with_dynamic_paths = type_with_dynamic_paths->createColumn();
+    auto & column_object_with_dynamic_paths = assert_cast<ColumnObject &>(*column_with_dynamic_paths);
+    column_object_with_dynamic_paths.insert(Object{{"a", Field{1u}}});
+    column_object_with_dynamic_paths.insert(Object{{"b", Field{1u}}});
+    column_object_with_dynamic_paths.insert(Object{{"c", Field{1u}}});
+    column_object_with_dynamic_paths.insert(Object{{"d", Field{1u}}});
+
+    auto type_with_shared_data_paths = DataTypeFactory::instance().get("JSON(max_dynamic_paths=0)");
+    auto column_with_shared_data_paths = type_with_shared_data_paths->createColumn();
+    auto & column_object_with_shared_data_paths = assert_cast<ColumnObject &>(*column_with_shared_data_paths);
+    column_object_with_shared_data_paths.insert(Object{});
+    column_object_with_shared_data_paths.insert(Object{{"a", Field{1u}}, {"c", Field{1u}}});
+    column_object_with_shared_data_paths.insert(Object{{"d", Field{1u}}, {"b", Field{1u}}});
+    column_object_with_shared_data_paths.insert(Object{});
+
+    std::unordered_map<String, MutableColumnPtr> dynamic_paths;
+    for (const auto & [path, column] : column_object_with_dynamic_paths.getDynamicPaths())
+        dynamic_paths[path] = IColumn::mutate(column);
+
+    auto column_object = ColumnObject::create({}, std::move(dynamic_paths), IColumn::mutate(column_object_with_shared_data_paths.getSharedDataPtr()), 4, 4, 16);
+    column_object->repairDuplicatesInDynamicPathsAndSharedData(0);
+    ASSERT_EQ((*column_object)[0], (Object{{"a", Field(1u)}}));
+    ASSERT_EQ((*column_object)[1], (Object{{"a", Field(1u)}, {"b", Field(1u)}, {"c", Field(1u)}}));
+    ASSERT_EQ((*column_object)[2], (Object{{"b", Field(1u)}, {"c", Field(1u)}, {"d", Field(1u)}}));
+    ASSERT_EQ((*column_object)[3], (Object{{"d", Field(1u)}}));
+}

src/DataTypes/Serializations/SerializationObject.cpp

@@ -1037,7 +1037,7 @@ void SerializationObject::deserializeBinaryBulkWithMultipleStreams(
     settings.path.pop_back();
     settings.path.pop_back();
 
-    column_object.validateDynamicPathsAndSharedData(shared_data_previous_size);
+    column_object.repairDuplicatesInDynamicPathsAndSharedData(shared_data_previous_size);
 }
 
 void SerializationObject::serializeBinary(const Field & field, WriteBuffer & ostr, const DB::FormatSettings & settings) const