Backport #75357 to 24.11: Fix crash in protobuf schema cache

Author: robot-clickhouse

src/Processors/Formats/Impl/ProtobufListInputFormat.cpp

@@ -19,12 +19,13 @@ ProtobufListInputFormat::ProtobufListInputFormat(
     const String & google_protos_path)
     : IRowInputFormat(header_, in_, params_)
     , reader(std::make_unique<ProtobufReader>(in_))
+    , descriptor_holder(ProtobufSchemas::instance().getMessageTypeForFormatSchema(
+          schema_info_.getSchemaInfo(), ProtobufSchemas::WithEnvelope::Yes, google_protos_path))
     , serializer(ProtobufSerializer::create(
           header_.getNames(),
           header_.getDataTypes(),
           missing_column_indices,
-          ProtobufSchemas::instance().getMessageTypeForFormatSchema(
-              schema_info_.getSchemaInfo(), ProtobufSchemas::WithEnvelope::Yes, google_protos_path),
+          descriptor_holder,
           /* with_length_delimiter = */ true,
           /* with_envelope = */ true,
           flatten_google_wrappers_,

src/Processors/Formats/Impl/ProtobufListInputFormat.h

@@ -6,6 +6,7 @@
 #    include <Formats/FormatSchemaInfo.h>
 #    include <Processors/Formats/IRowInputFormat.h>
 #    include <Processors/Formats/ISchemaReader.h>
+#    include <Formats/ProtobufSchemas.h>
 
 namespace DB
 {
@@ -44,6 +45,7 @@ class ProtobufListInputFormat final : public IRowInputFormat
 
     std::unique_ptr<ProtobufReader> reader;
     std::vector<size_t> missing_column_indices;
+    ProtobufSchemas::DescriptorHolder descriptor_holder;
     std::unique_ptr<ProtobufSerializer> serializer;
 };
 

src/Processors/Formats/Impl/ProtobufListOutputFormat.cpp

@@ -17,11 +17,12 @@ ProtobufListOutputFormat::ProtobufListOutputFormat(
     const String & google_protos_path)
     : IRowOutputFormat(header_, out_)
     , writer(std::make_unique<ProtobufWriter>(out))
+    , descriptor_holder(ProtobufSchemas::instance().getMessageTypeForFormatSchema(
+          schema_info_.getSchemaInfo(), ProtobufSchemas::WithEnvelope::Yes, google_protos_path))
     , serializer(ProtobufSerializer::create(
           header_.getNames(),
           header_.getDataTypes(),
-          ProtobufSchemas::instance().getMessageTypeForFormatSchema(
-              schema_info_.getSchemaInfo(), ProtobufSchemas::WithEnvelope::Yes, google_protos_path),
+          descriptor_holder,
           /* with_length_delimiter = */ true,
           /* with_envelope = */ true,
           defaults_for_nullable_google_wrappers_,

src/Processors/Formats/Impl/ProtobufListOutputFormat.h

@@ -4,7 +4,8 @@
 
 #if USE_PROTOBUF
 #    include <Processors/Formats/IRowOutputFormat.h>
-#   include <Formats/FormatSchemaInfo.h>
+#    include <Formats/FormatSchemaInfo.h>
+#    include <Formats/ProtobufSchemas.h>
 
 namespace DB
 {
@@ -42,6 +43,7 @@ class ProtobufListOutputFormat final : public IRowOutputFormat
     void resetFormatterImpl() override;
 
     std::unique_ptr<ProtobufWriter> writer;
+    ProtobufSchemas::DescriptorHolder descriptor_holder;
     std::unique_ptr<ProtobufSerializer> serializer;
 };
 

src/Processors/Formats/Impl/ProtobufRowInputFormat.h

@@ -55,9 +55,9 @@ class ProtobufRowInputFormat final : public IRowInputFormat
 
     std::unique_ptr<ProtobufReader> reader;
     std::vector<size_t> missing_column_indices;
+    const ProtobufSchemas::DescriptorHolder descriptor;
     std::unique_ptr<ProtobufSerializer> serializer;
 
-    const ProtobufSchemas::DescriptorHolder descriptor;
     bool with_length_delimiter;
     bool flatten_google_wrappers;
 };

src/Processors/Formats/Impl/ProtobufRowOutputFormat.cpp

@@ -24,11 +24,12 @@ ProtobufRowOutputFormat::ProtobufRowOutputFormat(
     bool with_length_delimiter_)
     : IRowOutputFormat(header_, out_)
     , writer(std::make_unique<ProtobufWriter>(out))
+    , descriptor_holder(ProtobufSchemas::instance().getMessageTypeForFormatSchema(
+          schema_info_.getSchemaInfo(), ProtobufSchemas::WithEnvelope::No, settings_.protobuf.google_protos_path))
     , serializer(ProtobufSerializer::create(
           header_.getNames(),
           header_.getDataTypes(),
-          ProtobufSchemas::instance().getMessageTypeForFormatSchema(
-              schema_info_.getSchemaInfo(), ProtobufSchemas::WithEnvelope::No, settings_.protobuf.google_protos_path),
+          descriptor_holder,
           with_length_delimiter_,
           /* with_envelope = */ false,
           settings_.protobuf.output_nullables_with_google_wrappers,

src/Processors/Formats/Impl/ProtobufRowOutputFormat.h

@@ -4,7 +4,8 @@
 
 #if USE_PROTOBUF
 #    include <Processors/Formats/IRowOutputFormat.h>
-#   include <Formats/FormatSchemaInfo.h>
+#    include <Formats/FormatSchemaInfo.h>
+#    include <Formats/ProtobufSchemas.h>
 
 namespace DB
 {
@@ -42,6 +43,7 @@ class ProtobufRowOutputFormat final : public IRowOutputFormat
     void writeField(const IColumn &, const ISerialization &, size_t) override {}
 
     std::unique_ptr<ProtobufWriter> writer;
+    ProtobufSchemas::DescriptorHolder descriptor_holder;
     std::unique_ptr<ProtobufSerializer> serializer;
     const bool allow_multiple_rows;
 };

tests/queries/0_stateless/03326_protobuf_schemas_race.reference

@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+# Tags: no-fasttest
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. "$CURDIR"/../shell_config.sh
+
+mkdir -p "${CLICKHOUSE_SCHEMA_FILES}"
+mkdir -p "${CLICKHOUSE_SCHEMA_FILES}/${CLICKHOUSE_TEST_UNIQUE_NAME}"
+SOURCE_SCHEMA_FILE="${CURDIR}/format_schemas/03234_proto_simple_nested_repeated_noexception.proto"
+TARGET_SCHEMA_FILE="${CLICKHOUSE_SCHEMA_FILES}/${CLICKHOUSE_TEST_UNIQUE_NAME}/03234_proto_simple_nested_repeated_noexception.proto"
+cp "${SOURCE_SCHEMA_FILE}" "${TARGET_SCHEMA_FILE}"
+
+echo "DROP TABLE IF EXISTS table_file;
+CREATE TABLE table_file (
+    u     UInt32,
+    \`v.w\`   Array(UInt32),
+    \`v.x\`   Array(UInt32),
+    \`v.y\`   Array(Array(UInt32)),
+    \`v.z\`   Array(Array(UInt32))
+) ENGINE File(Protobuf) SETTINGS format_schema = '$CLICKHOUSE_TEST_UNIQUE_NAME/03234_proto_simple_nested_repeated_noexception.proto:M';
+INSERT INTO table_file SELECT * FROM generateRandom() limit 1000000;
+DROP TABLE table_file;" | $CLICKHOUSE_CLIENT -m &
+
+for i in $(seq 1 100)
+do
+    $CLICKHOUSE_CLIENT -q "SYSTEM DROP FORMAT SCHEMA CACHE"
+done
+
+rm -rf "${CLICKHOUSE_SCHEMA_FILES}/${CLICKHOUSE_TEST_UNIQUE_NAME}"