Antalya 26.8: Backport of #117106 support dynamic type in parquet reader

src/Core/FormatFactorySettings.h

@@ -1732,6 +1732,9 @@ Indicate which field of protobuf oneof was found by means of setting enum value
 )", 0) \
     DECLARE(Bool, input_format_parquet_allow_geoparquet_parser, true, R"(
 Use geo column parser to convert Array(UInt8) into Point/MultiPoint/Linestring/Polygon/MultiLineString/MultiPolygon types
+)", 0) \
+    DECLARE(Bool, input_format_parquet_detect_variant_by_structure, true, R"(
+Read a Parquet group that has no `VARIANT` logical type, but has the layout of an unshredded variant (a `metadata` and a `value` field of type `BYTE_ARRAY`), as a variant, i.e. as `Dynamic`, instead of `Tuple`. Spark 4.0 writes variant columns this way. Groups annotated with the `VARIANT` logical type are always read as a variant.
 )", 0) \
     DECLARE(Bool, output_format_parquet_geometadata, true, R"(
 Allow to write information about geo columns in parquet metadata and encode columns in WKB format.

src/Core/SettingsChangesHistory.cpp

@@ -51,6 +51,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         });
         addSettingsChanges(settings_changes_history, "26.8",
         {
+            {"input_format_parquet_detect_variant_by_structure", false, true, "New setting to read Spark variant columns, which have no `VARIANT` logical type, as `Dynamic` instead of `Tuple`"},
             {"backup_s3_expired_token_retry_timeout_ms", 0, 120000, "Retry S3 backup and restore requests after an expired token; 0 restores the previous behavior."},
             {"allow_named_collection_override_by_default", true, true, "Obsolete. Overriding named collection keys requires `SHOW NAMED COLLECTIONS SECRETS`."},
             {"use_aggregation_memory_tracker", false, false, "New setting. In this release it is disabled by default, which restores the two-level conversion of `GROUP BY` from before 26.7 (https://github.com/ClickHouse/ClickHouse/pull/107490)."},

src/Formats/FormatFactory.cpp

@@ -265,6 +265,7 @@ FormatSettings getFormatSettings(const ContextPtr & context, const Settings & se
     format_settings.parquet.verify_checksums = settings[Setting::input_format_parquet_verify_checksums];
     format_settings.parquet.local_time_as_utc = settings[Setting::input_format_parquet_local_time_as_utc];
     format_settings.parquet.allow_geoparquet_parser = settings[Setting::input_format_parquet_allow_geoparquet_parser];
+    format_settings.parquet.detect_variant_by_structure = settings[Setting::input_format_parquet_detect_variant_by_structure];
     format_settings.parquet.write_geometadata = settings[Setting::output_format_parquet_geometadata];
     if (auto memory_limit = total_memory_tracker.getHardLimit(); memory_limit > 0)
     {

src/Formats/FormatSettings.h

@@ -420,6 +420,7 @@ struct FormatSettings
         double bloom_filter_bits_per_value = 10.5;
         size_t bloom_filter_flush_threshold_bytes = 1024 * 1024 * 128;
         bool allow_geoparquet_parser = true;
+        bool detect_variant_by_structure = true;
         bool spatial_filter_push_down = true;
         bool write_geometadata = true;
         size_t max_dictionary_size = 1024 * 1024;

src/Processors/Formats/Impl/Parquet/Reader.cpp

@@ -1,12 +1,14 @@
 #include <Common/logger_useful.h>
 #include <Common/ProfileEvents.h>
 #include <Columns/ColumnArray.h>
+#include <Columns/ColumnDynamic.h>
 #include <DataTypes/DataTypesNumber.h>
 #include <Processors/Formats/Impl/ArrowGeoTypes.h>
 #include <Columns/ColumnMap.h>
 #include <Columns/ColumnNullable.h>
 #include <Columns/ColumnsCommon.h>
 #include <Columns/ColumnString.h>
+#include <Columns/ColumnVariant.h>
 #include <Columns/ColumnsNumber.h>
 #include <Columns/ColumnTuple.h>
 #include <Columns/FilterDescription.h>
@@ -21,6 +23,7 @@
 #include <IO/Libdeflate.h>
 #include <Processors/Formats/Impl/Parquet/Decoding.h>
 #include <Processors/Formats/Impl/Parquet/GeoFilter.h>
+#include <Processors/Formats/Impl/Parquet/VariantEncoding.h>
 #include <Processors/Formats/Impl/Parquet/parquetBloomFilterHash.h>
 #include <Processors/Formats/Impl/Parquet/Reader.h>
 #include <Processors/Formats/Impl/Parquet/SchemaConverter.h>
@@ -3295,9 +3298,8 @@ MutableColumnPtr Reader::formOutputColumn(RowSubgroup & row_subgroup, size_t out
             nullable_group_null_map = ColumnUInt8::create(num_rows, UInt8(0));
     }
 
-    TypeIndex kind = output_info.nullable_group
-        ? removeNullable(output_info.input_type)->getColumnType()
-        : output_info.input_type->getColumnType();
+    /// Nullable wraps the type of a physically nullable tuple group and of a variant read as `Nullable(JSON)`.
+    TypeIndex kind = removeNullable(output_info.input_type)->getColumnType();
 
     if (output_info.is_primitive)
     {
@@ -3349,6 +3351,16 @@ MutableColumnPtr Reader::formOutputColumn(RowSubgroup & row_subgroup, size_t out
         else
             res = ColumnTuple::create(std::move(columns));
     }
+    else if (kind == TypeIndex::Dynamic || kind == TypeIndex::Object)
+    {
+        chassert(output_info.nested_columns.size() == 2);
+        MutableColumnPtr metadata = formOutputColumn(row_subgroup, output_info.nested_columns[0], num_rows);
+        MutableColumnPtr value = formOutputColumn(row_subgroup, output_info.nested_columns[1], num_rows);
+
+        res = output_info.input_type->createColumn();
+        res->reserve(num_rows);
+        decodeVariantColumn(*metadata, *value, *res, output_info.input_type, output_info.name, num_rows, options.format);
+    }
     else
     {
         chassert(kind == TypeIndex::Map);

src/Processors/Formats/Impl/Parquet/SchemaConverter.cpp

@@ -1,3 +1,4 @@
+#include <DataTypes/DataTypeDynamic.h>
 #include <Processors/Formats/Impl/Parquet/SchemaConverter.h>
 
 #include <Common/checkStackSize.h>
@@ -312,7 +313,8 @@ void SchemaConverter::processSubtree(TraversalNode & node)
     if (!processSubtreePrimitive(node) &&
         !processSubtreeMap(node) &&
         !processSubtreeArrayOuter(node) &&
-        !processSubtreeArrayInner(node))
+        !processSubtreeArrayInner(node) &&
+        !processSubtreeDynamic(node))
     {
         processSubtreeTuple(node);
     }
@@ -684,6 +686,149 @@ static bool tupleSubtreeIsAllRequired(const std::vector<parq::SchemaElement> & s
     return true;
 }
 
+static std::optional<size_t> subtreeEnd(const std::vector<parq::SchemaElement> & schema, size_t idx)
+{
+    size_t remaining = 1;
+    while (remaining > 0)
+    {
+        if (idx >= schema.size() || remaining > schema.size())
+            return std::nullopt;
+        const parq::SchemaElement & element = schema[idx];
+        remaining -= 1;
+        if (!isPrimitiveNode(element))
+        {
+            if (element.num_children < 0 || size_t(element.num_children) > schema.size())
+                return std::nullopt;
+            remaining += size_t(element.num_children);
+        }
+        idx += 1;
+    }
+    return idx;
+}
+
+bool SchemaConverter::processSubtreeDynamic(TraversalNode & node)
+{
+    if (node.element->num_children != 2 && node.element->num_children != 3)
+        return false;
+    const size_t num_children = size_t(node.element->num_children);
+    if (num_children > file_metadata.schema.size() - schema_idx)
+        return false;
+
+    const bool group_annotated_variant = node.element->logicalType.__isset.VARIANT;
+
+    enum ChildRole : size_t { Metadata = 0, Value = 1, TypedValue = 2, NumRoles = 3 };
+
+    std::array<std::optional<size_t>, NumRoles> schema_idx_of_role;
+    std::vector<size_t> role_of_child(num_children);
+    size_t child_schema_idx = schema_idx;
+    for (size_t i = 0; i < num_children; ++i)
+    {
+        const parq::SchemaElement & child = file_metadata.schema[child_schema_idx];
+        size_t role = 0;
+        if (child.name == "metadata")
+            role = Metadata;
+        else if (child.name == "value")
+            role = Value;
+        else if (child.name == "typed_value")
+            role = TypedValue;
+        else
+            return false;
+
+        if (schema_idx_of_role[role].has_value())
+            return false;
+        schema_idx_of_role[role] = child_schema_idx;
+        role_of_child[i] = role;
+
+        std::optional<size_t> end = subtreeEnd(file_metadata.schema, child_schema_idx);
+        if (!end.has_value())
+            return false;
+        child_schema_idx = *end;
+    }
+
+    if (!schema_idx_of_role[Metadata].has_value())
+        return false;
+    if (!schema_idx_of_role[Value].has_value() && !schema_idx_of_role[TypedValue].has_value())
+        return false;
+
+    /// `metadata` and `value` hold variant-encoded blobs: unannotated BYTE_ARRAY, not an array.
+    auto is_variant_blob_leaf = [&](size_t idx)
+    {
+        const parq::SchemaElement & element = file_metadata.schema[idx];
+        return isPrimitiveNode(element)
+            && element.type == parq::Type::BYTE_ARRAY
+            && element.repetition_type != parq::FieldRepetitionType::REPEATED
+            && (group_annotated_variant || (!element.logicalType.__isset.STRING && !element.__isset.converted_type));
+    };
+
+    if (!is_variant_blob_leaf(*schema_idx_of_role[Metadata]))
+        return false;
+    if (schema_idx_of_role[Value].has_value() && !is_variant_blob_leaf(*schema_idx_of_role[Value]))
+        return false;
+
+    const DataTypePtr type_hint = node.type_hint ? removeNullable(node.type_hint) : nullptr;
+    const bool read_as_json = type_hint && isObject(type_hint);
+    if (type_hint && !read_as_json && !isDynamic(type_hint))
+        return false;
+
+    /// Spark 4.0 writes variant columns without the `VARIANT` logical type, so they can only be
+    /// recognized by structure. Before this was supported, such groups were read as `Tuple`, hence
+    /// the setting. An explicit `Dynamic`/`JSON` request is not ambiguous and bypasses it.
+    if (!group_annotated_variant && !node.type_hint && !options.format.parquet.detect_variant_by_structure)
+        return false;
+
+    std::optional<String> unsupported;
+    if (group_annotated_variant && node.element->logicalType.VARIANT.__isset.specification_version
+        && node.element->logicalType.VARIANT.specification_version != 1)
+        unsupported = fmt::format(
+            "a variant with specification version {}, but only version 1 is supported",
+            Int16(node.element->logicalType.VARIANT.specification_version));
+    else if (schema_idx_of_role[TypedValue].has_value())
+        unsupported = "a shredded variant (it has a `typed_value` field), which is not supported yet";
+
+    size_t primitive_start = primitive_columns.size();
+    size_t output_start = output_columns.size();
+
+    /// Recurse even when the column is unsupported or not requested, to advance the traversal
+    /// past the group's children.
+    std::array<std::optional<size_t>, NumRoles> output_idx_of_role;
+    for (size_t i = 0; i < num_children; ++i)
+    {
+        TraversalNode subnode = node.prepareToRecurse(SchemaContext::None, nullptr);
+        subnode.requested = node.requested && !unsupported.has_value();
+        processSubtree(subnode);
+        output_idx_of_role[role_of_child[i]] = subnode.output_idx;
+    }
+
+    if (!node.requested)
+        return true;
+
+    if (unsupported.has_value())
+    {
+        /// Like for unsupported primitive types, schema inference can skip the column. When reading,
+        /// a requested column must not silently come out empty.
+        if (!sample_block && options.format.parquet.skip_columns_with_unsupported_types_in_schema_inference)
+            return true;
+        throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Parquet column {} is {}", node.getNameForLogging(), *unsupported);
+    }
+
+    if (!output_idx_of_role[Metadata].has_value() || !output_idx_of_role[Value].has_value())
+    {
+        primitive_columns.resize(primitive_start);
+        output_columns.resize(output_start);
+        return true;
+    }
+
+    node.output_idx = output_columns.size();
+    OutputColumnInfo & output = output_columns.emplace_back();
+    output.name = node.name;
+    output.primitive_start = primitive_start;
+    output.primitive_end = primitive_columns.size();
+    output.input_type = read_as_json ? node.type_hint : std::make_shared<DataTypeDynamic>();
+    output.output_type = output.input_type;
+    output.nested_columns = {output_idx_of_role[Metadata].value(), output_idx_of_role[Value].value()};
+    return true;
+}
+
 void SchemaConverter::processSubtreeTuple(TraversalNode & node)
 {
     /// Tuple (possibly a Map key_value tuple):

src/Processors/Formats/Impl/Parquet/SchemaConverter.h

@@ -144,6 +144,7 @@ struct SchemaConverter
     bool processSubtreeArrayOuter(TraversalNode & node);
     bool processSubtreeArrayInner(TraversalNode & node);
     void processSubtreeTuple(TraversalNode & node);
+    bool processSubtreeDynamic(TraversalNode & node);
 
     void processPrimitiveColumn(
         const parq::SchemaElement & element, DataTypePtr type_hint,