Hello,

after updating clickhouse from 21.2.3.15-2 to 21.3.2.5-2 I see those warnings :

 `<Warning> default.osb4_reverse.DirectoryMonitor: Structure does not match (remote: Path LowCardinality(String) ColumnLowCardinality(size = 0, UInt8(size = 0), ColumnUnique(size = 1, String(size = 1))), Value Float64 Float64(size = 0), Time UInt32 UInt32(size = 0), Date Date UInt16(size = 0), Timestamp UInt32 UInt32(size = 0), local: Path LowCardinality(String) ColumnLowCardinality(size = 0, UInt16(size = 0), ColumnUnique(size = 1, String(size = 1))), Value Float64 Float64(size = 0), Time UInt32 UInt32(size = 0), Date Date UInt16(size = 0), Timestamp UInt32 UInt32(size = 0)), implicit conversion will be done`

I think this is related to https://github.com/ClickHouse/ClickHouse/pull/19947

The only difference between remote and local I found inside this log is :

ColumnLowCardinality(size = 0, UInt8(size = 0) to
ColumnLowCardinality(size = 0, UInt16(size = 0)

But I don't see the difference between the tables.

```
SHOW CREATE TABLE default.osb4_reverse

CREATE TABLE default.osb4_reverse
(
    `Path` LowCardinality(String) CODEC(ZSTD(15)),
    `Value` Float64 CODEC(Gorilla, ZSTD(22)),
    `Time` UInt32 CODEC(DoubleDelta, ZSTD(15)),
    `Date` Date CODEC(Delta(2), ZSTD(22)),
    `Timestamp` UInt32 CODEC(ZSTD(15))
)
ENGINE = Distributed('osb4_hpc_metrics', '', 'osb4_reverse', rand())


CREATE TABLE osb4_metrics_r0.osb4_reverse
(
    `Path` LowCardinality(String) CODEC(ZSTD(15)),
    `Value` Float64 CODEC(Gorilla, ZSTD(22)),
    `Time` UInt32 CODEC(DoubleDelta, ZSTD(15)),
    `Date` Date CODEC(Delta(2), ZSTD(22)),
    `Timestamp` UInt32 CODEC(ZSTD(15))
)
ENGINE = ReplicatedReplacingMergeTree('/clickhouse/clusters/osb4_hpc_metrics/tables/{table4}/data', '{host2}')
PARTITION BY toYYYYMM(Date)
ORDER BY (Path, Time)
TTL Date + toIntervalMonth(6)
SETTINGS index_granularity = 8192
```

Any hints ?

Ralph
writeToLocal(block, shard_info.getLocalNodeCount());
        std::vector<std::string> dir_names;
        for (const auto & address : cluster->getShardsAddresses()[shard_id])
            if (!address.is_local || !settings.prefer_localhost_replica)
                dir_names.push_back(address.toFullString(settings.use_compact_format_in_distributed_parts_names));
        if (!dir_names.empty())
            writeToShard(block, dir_names);
    }
}
void DistributedBlockOutputStream::writeToLocal(const Block & block, size_t repeats)
{
    
    InterpreterInsertQuery interp(query_ast, context);
    auto block_io = interp.execute();
    block_io.out->writePrefix();
    writeBlockConvert(block_io.out, block, repeats, log);
    block_io.out->writeSuffix();
}
void DistributedBlockOutputStream::writeToShard(const Block & block, const std::vector<std::string> & dir_names)
{
    const auto & settings = context.getSettingsRef();
    const auto & distributed_settings = storage.getDistributedSettingsRef();
    bool fsync = distributed_settings.fsync_after_insert;
    bool dir_fsync = distributed_settings.fsync_directories;
    std::string compression_method = Poco::toUpper(settings.network_compression_method.toString());
    std::optional<int> compression_level;
    if (compression_method == "ZSTD")
        compression_level = settings.network_zstd_compression_level;
    CompressionCodecFactory::instance().validateCodec(compression_method, compression_level, !settings.allow_suspicious_codecs);
    CompressionCodecPtr compression_codec = CompressionCodecFactory::instance().get(compression_method, compression_level);