Fix deserialization of variant discriminators with basic format in compact parts

Author: Avogar

src/DataTypes/Serializations/SerializationVariant.cpp

@@ -452,12 +452,15 @@ void SerializationVariant::deserializeBinaryBulkWithMultipleStreams(
     {
         variant_state = checkAndGetState<DeserializeBinaryBulkStateVariant>(state);
         auto * discriminators_state = checkAndGetState<DeserializeBinaryBulkStateVariantDiscriminators>(variant_state->discriminators_state);
+        /// Now we are sure that discriminators are not in cache and we can save the size of discriminators now to know how
+        /// many discriminators were actually deserialized to iterate over them later to calculate limits for variants.
+        variant_state->num_rows_read = col.getLocalDiscriminatorsPtr()->size();
 
         /// Deserialize discriminators according to serialization mode.
         if (discriminators_state->mode.value == DiscriminatorsSerializationMode::BASIC)
         {
             SerializationNumber<ColumnVariant::Discriminator>().deserializeBinaryBulk(
-                *col.getLocalDiscriminatorsPtr()->assumeMutable(), *discriminators_stream, 0, rows_offset + limit, 0);
+                *col.getLocalDiscriminatorsPtr()->assumeMutable(), *discriminators_stream, rows_offset, limit, 0);
         }
         else
         {
@@ -738,13 +741,17 @@ void SerializationVariant::deserializeBinary(IColumn & column, ReadBuffer & istr
     {
         col.insertDefault();
     }
-    else
+    else if (global_discr < variants.size())
     {
         auto & variant_column = col.getVariantByGlobalDiscriminator(global_discr);
         variants[global_discr]->deserializeBinary(variant_column, istr, settings);
         col.getLocalDiscriminators().push_back(col.localDiscriminatorByGlobal(global_discr));
         col.getOffsets().push_back(variant_column.size() - 1);
     }
+    else
+    {
+        throw Exception(ErrorCodes::INCORRECT_DATA, "Cannot read value of {}: unexpected discriminator {}", variant_name, UInt64(global_discr));
+    }
 }
 
 namespace

tests/queries/0_stateless/03411_variant_basic_discriminators_deserialization_bug.reference

@@ -0,0 +1,2 @@
+42
+a

tests/queries/0_stateless/03411_variant_basic_discriminators_deserialization_bug.sql

@@ -0,0 +1,8 @@
+set max_threads=1;
+DROP TABLE IF EXISTS t0;
+CREATE TABLE t0 (c0 Variant(String, Int)) ENGINE = MergeTree() PRIMARY KEY tuple() SETTINGS use_compact_variant_discriminators_serialization = 0, index_granularity=1;
+INSERT INTO TABLE t0 (c0) VALUES (42), ('a');
+optimize table t0 final;
+SELECT c0 FROM t0;
+DROP TABLE t0;
+