```sql
CREATE TABLE t
(
   id Int64, 
   d String,
   p Map(String, String)
   
)
ENGINE = ReplacingMergeTree order by id settings index_granularity  = 0;


insert into t values (1,  's', {'a':'b'});

CH starts eating RAM until it crashes

the same with

insert into t select 1,  's', map('a','b');

```

const CompressionCodecPtr & default_codec_,
    const MergeTreeWriterSettings & settings_,
    const MergeTreeIndexGranularity & index_granularity_)
    : MergeTreeDataPartWriterOnDisk(data_part_, columns_list_, metadata_snapshot_,
        indices_to_recalc_, marks_file_extension_,
        default_codec_, settings_, index_granularity_)
    , plain_file(data_part_->getDataPartStorage().writeFile(
            MergeTreeDataPartCompact::DATA_FILE_NAME_WITH_EXTENSION,
            settings.max_compress_block_size,
            settings_.query_write_settings))
    , plain_hashing(*plain_file)
{
    marks_file = data_part_->getDataPartStorage().writeFile(
            MergeTreeDataPartCompact::DATA_FILE_NAME + marks_file_extension_,
            4096,
            settings_.query_write_settings);
    marks_file_hashing = std::make_unique<HashingWriteBuffer>(*marks_file);
    if (data_part_->index_granularity_info.mark_type.compressed)
    {
        marks_compressor = std::make_unique<CompressedWriteBuffer>(
            *marks_file_hashing,
            settings_.getMarksCompressionCodec(),
            settings_.marks_compress_block_size);
        marks_source_hashing = std::make_unique<HashingWriteBuffer>(*marks_compressor);
    }
    const auto & storage_columns = metadata_snapshot->getColumns();
    for (const auto & column : columns_list)
        addStreams(column, storage_columns.getCodecDescOrDefault(column.name, default_codec));
}
void MergeTreeDataPartWriterCompact::addStreams(const NameAndTypePair & column, const ASTPtr & effective_codec_desc)