Some intermediate stage 2

src/Compression/CompressionFactory.h

@@ -38,7 +38,7 @@ public:
     CompressionCodecPtr getDefaultCodec() const;
 
     /// Validate codecs AST specified by user and parses codecs description (substitute default parameters)
-    ASTPtr validateCodecAndGetPreprocessedAST(const ASTPtr & ast, DataTypePtr column_type, bool sanity_check) const;
+    ASTPtr validateCodecAndGetPreprocessedAST(const ASTPtr & ast, const IDataType * column_type, bool sanity_check) const;
 
     /// Validate codecs AST specified by user
     void validateCodec(const String & family_name, std::optional<int> level, bool sanity_check) const;
@@ -48,7 +48,7 @@ public:
     /// be able to work without information about type. Also AST can contain
     /// codec, which can be alias to current default codec, which can be changed
     /// in runtime.
-    CompressionCodecPtr get(const ASTPtr & ast, DataTypePtr column_type, CompressionCodecPtr current_default = nullptr) const;
+    CompressionCodecPtr get(const ASTPtr & ast, const IDataType * column_type, CompressionCodecPtr current_default = nullptr) const;
 
     /// Get codec by method byte (no params available)
     CompressionCodecPtr get(const uint8_t byte_code) const;

src/DataTypes/IDataType.cpp

@@ -130,6 +130,18 @@ String IDataType::getFileNameForStream(const String & column_name, const IDataTy
 }
 
 
+bool IDataType::isNonGenericCompressionAllowedForStream(const SubstreamPath & path)
+{
+    for (const Substream & elem : path)
+    {
+        if (elem.type == Substream::NullMap
+            || elem.type == Substream::ArraySizes
+            || elem.type == Substream::DictionaryIndexes)
+            return false;
+    }
+    return true;
+}
+
 void IDataType::insertDefaultInto(IColumn & column) const
 {
     column.insertDefault();

src/Storages/ColumnsDescription.cpp

@@ -426,6 +426,16 @@ CompressionCodecPtr ColumnsDescription::getCodecOrDefault(const String & column_
     return getCodecOrDefault(column_name, CompressionCodecFactory::instance().getDefaultCodec());
 }
 
+ASTPtr ColumnsDescription::getCodecDescOrDefault(const String & column_name, CompressionCodecPtr default_codec)
+{
+    const auto it = columns.get<1>().find(column_name);
+
+    if (it == columns.get<1>().end() || !it->codec)
+        return default_codec->getFullCodecDesc();
+
+    return it->codec;
+}
+
 ColumnsDescription::ColumnTTLs ColumnsDescription::getColumnTTLs() const
 {
     ColumnTTLs ret;

src/Storages/ColumnsDescription.h

@@ -115,6 +115,7 @@ public:
     bool hasCompressionCodec(const String & column_name) const;
     CompressionCodecPtr getCodecOrDefault(const String & column_name, CompressionCodecPtr default_codec) const;
     CompressionCodecPtr getCodecOrDefault(const String & column_name) const;
+    ASTPtr getCodecDescOrDefault(const String & column_name, CompressionCodecPtr default_codec);
 
     String toString() const;
     static ColumnsDescription parse(const String & str);

src/Storages/MergeTree/MergeTreeDataPartWriterWide.h

@@ -66,7 +66,8 @@ private:
     void addStreams(
         const String & name,
         const IDataType & type,
-        const CompressionCodecPtr & effective_codec,
+        const ASTPtr & effective_codec_desc,
+        const CompressionCodecPtr & default_codec,
         size_t estimated_size);
 
     SerializationStates serialization_states;