Merge: Cast types on read [#CLICKHOUSE-2957]

dbms/src/DataStreams/CastEnumBlockInputStream.cpp

@@ -10,7 +10,7 @@ namespace DB
 {
 
 CastEnumBlockInputStream::CastEnumBlockInputStream(
-    Context & context_,
+    const Context & context_,
     BlockInputStreamPtr input_,
     const Block & in_sample_,
     const Block & out_sample_)
@@ -109,12 +109,18 @@ void CastEnumBlockInputStream::collectEnums(const Block & in_sample, const Block
         const auto & in_elem  = in_sample.getByPosition(i);
         const auto & out_elem = out_sample.getByPosition(i);
 
-        /// Force conversion only if source type is not Enum.
+        /// Force conversion if source type is not Enum.
         if ( dynamic_cast<IDataTypeEnum*>(out_elem.type.get()) &&
             !dynamic_cast<IDataTypeEnum*>(in_elem.type.get()))
         {
             enum_types[i] = NameAndTypePair(out_elem.name, out_elem.type);
         }
+
+        /// Force conversion if both types is numeric but not equal.
+        if (in_elem.type->behavesAsNumber() && out_elem.type->behavesAsNumber() && !out_elem.type->equals(*in_elem.type))
+        {
+            enum_types[i] = NameAndTypePair(out_elem.name, out_elem.type);
+        }
     }
 }
 

dbms/src/DataStreams/CastEnumBlockInputStream.h

@@ -15,7 +15,7 @@ class IFunction;
 class CastEnumBlockInputStream : public IProfilingBlockInputStream
 {
 public:
-    CastEnumBlockInputStream(Context & context_,
+    CastEnumBlockInputStream(const Context & context_,
                              BlockInputStreamPtr input_,
                              const Block & in_sample_,
                              const Block & out_sample_);
@@ -31,7 +31,7 @@ private:
     void collectEnums(const Block & in_sample, const Block & out_sample);
 
 private:
-    Context & context;
+    const Context & context;
     std::vector<std::experimental::optional<NameAndTypePair>> enum_types;
     std::vector<std::shared_ptr<IFunction>> cast_functions;  /// Used to perform type conversions.
 };

dbms/src/Storages/StorageMerge.cpp

@@ -11,6 +11,7 @@
 #include <DataTypes/DataTypeString.h>
 #include <Columns/ColumnString.h>
 #include <Databases/IDatabase.h>
+#include <DataStreams/CastEnumBlockInputStream.h>
 
 
 namespace DB
@@ -176,6 +177,9 @@ BlockInputStreams StorageMerge::read(
             else if (processed_stage_in_source_table != processed_stage_in_source_tables.value())
                 throw Exception("Source tables for Merge table are processing data up to different stages",
                     ErrorCodes::INCOMPATIBLE_SOURCE_TABLES);
+
+            for (auto & stream : source_streams)
+                stream = std::make_shared<CastEnumBlockInputStream>(context, stream, table->getSampleBlock(), getSampleBlock());
         }
         else
         {
@@ -199,7 +203,10 @@ BlockInputStreams StorageMerge::read(
                     throw Exception("Source tables for Merge table are processing data up to different stages",
                         ErrorCodes::INCOMPATIBLE_SOURCE_TABLES);
 
-                return streams.empty() ? std::make_shared<NullBlockInputStream>() : streams.front();
+                auto stream = streams.empty() ? std::make_shared<NullBlockInputStream>() : streams.front();
+                if (!streams.empty())
+                    stream = std::make_shared<CastEnumBlockInputStream>(context, stream, table->getSampleBlock(), getSampleBlock());
+                return stream;
             }));
         }
 

dbms/tests/queries/0_stateless/00458_merge_type_cast.reference

+ = 1:
+1
+1
+ 1:
+1
+1
+ 4294967290:
+4294967290
+4294967290
+ 4294967299:
+4294967299
+ 1:
+1
+1
+ -1: 

dbms/tests/queries/0_stateless/00458_merge_type_cast.sql

+DROP TABLE IF EXISTS test.u32;
+DROP TABLE IF EXISTS test.u64;
+DROP TABLE IF EXISTS test.merge_32_64;
+
+CREATE TABLE test.u32 (x UInt32) ENGINE = Memory;
+CREATE TABLE test.u64 (x UInt64) ENGINE = Memory;
+CREATE TABLE test.merge_32_64 (x UInt64) ENGINE = Merge(test, 'u32|u64');
+
+INSERT INTO test.u32 VALUES (1);
+INSERT INTO test.u64 VALUES (1);
+
+INSERT INTO test.u32 VALUES (4294967290);
+INSERT INTO test.u64 VALUES (4294967290);
+--now inserts 3. maybe need out of range check?
+--INSERT INTO test.u32 VALUES (4294967299);
+INSERT INTO test.u64 VALUES (4294967299);
+
+select ' = 1:';
+SELECT x FROM test.merge_32_64 WHERE x = 1;
+select ' 1:';
+SELECT x FROM test.merge_32_64 WHERE x IN (1);
+select ' 4294967290:';
+SELECT x FROM test.merge_32_64 WHERE x IN (4294967290);
+select ' 4294967299:';
+SELECT x FROM test.merge_32_64 WHERE x IN (4294967299);
+--select ' -1: ';
+--SELECT x FROM test.merge_32_64 WHERE x IN (-1);
+
+DROP TABLE test.u32;
+DROP TABLE test.u64;
+DROP TABLE test.merge_32_64;
+
+
+
+DROP TABLE IF EXISTS test.s64;
+DROP TABLE IF EXISTS test.u64;
+DROP TABLE IF EXISTS test.merge_s64_u64;
+
+CREATE TABLE test.s64 (x Int64) ENGINE = Memory;
+CREATE TABLE test.u64 (x UInt64) ENGINE = Memory;
+CREATE TABLE test.merge_s64_u64 (x UInt64) ENGINE = Merge(test, 's64|u64');
+
+INSERT INTO test.s64 VALUES (1);
+INSERT INTO test.s64 VALUES (-1);
+INSERT INTO test.u64 VALUES (1);
+
+select ' 1:';
+SELECT x FROM test.merge_s64_u64 WHERE x IN (1);
+select ' -1: ';
+SELECT x FROM test.merge_s64_u64 WHERE x IN (-1);
+
+DROP TABLE test.s64;
+DROP TABLE test.u64;
+DROP TABLE test.merge_s64_u64;