Support for 'time' type in Iceberg

src/Processors/Formats/Impl/AvroRowInputFormat.cpp

@@ -159,6 +159,9 @@ static void insertNumber(IColumn & column, WhichDataType type, T value)
         case TypeIndex::DateTime64:
             assert_cast<ColumnDecimal<DateTime64> &>(column).insertValue(static_cast<Int64>(value));
             break;
+        case TypeIndex::Time64:
+            assert_cast<ColumnDecimal<Time64> &>(column).insertValue(static_cast<Int64>(value));
+            break;
         case TypeIndex::IPv4:
             assert_cast<ColumnIPv4 &>(column).insertValue(IPv4(static_cast<UInt32>(value)));
             break;
@@ -307,6 +310,8 @@ AvroDeserializer::DeserializeFn AvroDeserializer::createDeserializeFn(const avro
                 return createDecimalDeserializeFn<DataTypeDecimal256>(root_node, target_type, false);
             if (target.isDateTime64())
                 return createDecimalDeserializeFn<DataTypeDateTime64>(root_node, target_type, false);
+            if (target.isTime64())
+                return createDecimalDeserializeFn<DataTypeTime64>(root_node, target_type, false);
             break;
         case avro::AVRO_INT:
             if (target_type->isValueRepresentedByNumber())
@@ -1300,8 +1305,11 @@ DataTypePtr AvroSchemaReader::avroNodeToDataType(avro::NodePtr node)
     {
         case avro::Type::AVRO_INT:
         {
-            if (node->logicalType().type() == avro::LogicalType::DATE)
+            auto logical_type = node->logicalType();
+            if (logical_type.type() == avro::LogicalType::DATE)
                 return {std::make_shared<DataTypeDate32>()};
+            if (logical_type.type() == avro::LogicalType::TIME_MILLIS)
+                return {std::make_shared<DataTypeTime64>(3)};
 
             return {std::make_shared<DataTypeInt32>()};
         }
@@ -1312,6 +1320,10 @@ DataTypePtr AvroSchemaReader::avroNodeToDataType(avro::NodePtr node)
                 return {std::make_shared<DataTypeDateTime64>(3)};
             if (logical_type.type() == avro::LogicalType::TIMESTAMP_MICROS)
                 return {std::make_shared<DataTypeDateTime64>(6)};
+            if (logical_type.type() == avro::LogicalType::TIME_MILLIS)
+                return {std::make_shared<DataTypeTime64>(3)};
+            if (logical_type.type() == avro::LogicalType::TIME_MICROS)
+                return {std::make_shared<DataTypeTime64>(6)};
 
             return std::make_shared<DataTypeInt64>();
         }

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

@@ -21,6 +21,7 @@
 #include <DataTypes/DataTypeMap.h>
 #include <DataTypes/DataTypeDateTime64.h>
 #include <DataTypes/DataTypeFixedString.h>
+#include <DataTypes/DataTypeTime64.h>
 #include <DataTypes/DataTypeCustom.h>
 
 /// This file deals with schema conversion and with repetition and definition levels.
@@ -475,6 +476,55 @@ void preparePrimitiveColumn(ColumnPtr column, DataTypePtr type, const std::strin
         case TypeIndex::Decimal128: decimal(16, getDecimalPrecision(*type), getDecimalScale(*type)); break;
         case TypeIndex::Decimal256: decimal(32, getDecimalPrecision(*type), getDecimalScale(*type)); break;
 
+        case TypeIndex::Time:
+        {
+            parq::TimeUnit unit;
+            unit.__set_MICROS({});
+
+            parq::TimeType tt;
+            tt.__set_isAdjustedToUTC(false);
+            tt.__set_unit(unit);
+
+            parq::LogicalType t;
+            t.__set_TIME(tt);
+            types(T::INT64, parq::ConvertedType::TIME_MICROS, t);
+            state.datetime_multiplier = 1'000'000;
+            break;
+        }
+
+        case TypeIndex::Time64:
+        {
+            std::optional<parq::ConvertedType::type> converted;
+            parq::TimeUnit unit;
+            const auto & dt = assert_cast<const DataTypeTime64 &>(*type);
+            UInt32 scale = dt.getScale();
+            UInt32 converted_scale;
+            if (scale <= 6)
+            {
+                converted = parq::ConvertedType::TIME_MICROS;
+                unit.__set_MICROS({});
+                converted_scale = 6;
+            }
+            else if (scale <= 9)
+            {
+                unit.__set_NANOS({});
+                converted_scale = 9;
+            }
+            else
+            {
+                throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected Time64 scale: {}", scale);
+            }
+
+            parq::TimeType tt;
+            tt.__set_isAdjustedToUTC(false);
+            tt.__set_unit(unit);
+            parq::LogicalType t;
+            t.__set_TIME(tt);
+            types(T::INT64, converted, t);
+            state.datetime_multiplier = DataTypeTime64::getScaleMultiplier(converted_scale - scale);
+            break;
+        }
+
         default:
             throw Exception(ErrorCodes::UNKNOWN_TYPE, "Internal type '{}' of column '{}' is not supported for conversion into Parquet data format.", type->getFamilyName(), name);
     }

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

@@ -360,16 +360,17 @@ struct ConverterNumeric
     }
 };
 
-struct ConverterDateTime64WithMultiplier
+template <typename TYPE>
+struct ConverterTimeType64WithMultiplierImpl
 {
     using Statistics = StatisticsNumeric<Int64, Int64>;
 
-    using Col = ColumnDecimal<DateTime64>;
+    using Col = ColumnDecimal<TYPE>;
     const Col & column;
     Int64 multiplier;
     PODArray<Int64> buf;
 
-    ConverterDateTime64WithMultiplier(const ColumnPtr & c, Int64 multiplier_) : column(assert_cast<const Col &>(*c)), multiplier(multiplier_) {}
+    ConverterTimeType64WithMultiplierImpl(const ColumnPtr & c, Int64 multiplier_) : column(assert_cast<const Col &>(*c)), multiplier(multiplier_) {}
 
     const Int64 * getBatch(size_t offset, size_t count)
     {
@@ -387,6 +388,9 @@ struct ConverterDateTime64WithMultiplier
     }
 };
 
+using ConverterDateTime64WithMultiplier = ConverterTimeType64WithMultiplierImpl<DateTime64>;
+using ConverterTime64WithMultiplier = ConverterTimeType64WithMultiplierImpl<Time64>;
+
 /// Multiply DateTime by 1000 to get milliseconds (because Parquet doesn't support seconds).
 struct ConverterDateTime
 {
@@ -407,6 +411,26 @@ struct ConverterDateTime
     }
 };
 
+struct ConverterTime
+{
+    using Statistics = StatisticsNumeric<Int64, Int64>;
+
+    using Col = ColumnVector<Int32>;
+    const Col & column;
+    PODArray<Int64> buf;
+    Int64 multiplier;
+
+    ConverterTime(const ColumnPtr & c, Int64 multiplier_) : column(assert_cast<const Col &>(*c)), multiplier(multiplier_) {}
+
+    const Int64 * getBatch(size_t offset, size_t count)
+    {
+        buf.resize(count);
+        for (size_t i = 0; i < count; ++i)
+            buf[i] = static_cast<Int64>(column.getData()[offset + i]) * multiplier;
+        return buf.data();
+    }
+};
+
 struct ConverterString
 {
     using Statistics = StatisticsStringRef;
@@ -1171,7 +1195,12 @@ void writeColumnChunkBody(
                 N(Int16, Int32Type);
             break;
         }
-        case TypeIndex::Int32  : N(Int32,  Int32Type); break;
+        case TypeIndex::Int32:
+            if (s.type->getTypeId() == TypeIndex::Time)
+                writeColumnImpl<parquet::Int64Type>(s, options, out, ConverterTime(s.primitive_column, s.datetime_multiplier));
+            else
+                N(Int32,  Int32Type);
+            break;
         case TypeIndex::Int64  : N(Int64,  Int64Type); break;
 
         case TypeIndex::UInt32:
@@ -1184,7 +1213,6 @@ void writeColumnChunkBody(
                 writeColumnImpl<parquet::Int64Type>(s, options, out, ConverterDateTime(s.primitive_column));
             }
             break;
-
         #undef N
 
         case TypeIndex::Float32:
@@ -1253,6 +1281,17 @@ void writeColumnChunkBody(
         case TypeIndex::Decimal256: D(Decimal256); break;
         #undef D
 
+        case TypeIndex::Time64:
+            if (s.datetime_multiplier == 1)
+                writeColumnImpl<parquet::Int64Type>(
+                    s, options, out, ConverterNumeric<ColumnDecimal<Time64>, Int64, Int64>(
+                        s.primitive_column));
+            else
+                writeColumnImpl<parquet::Int64Type>(
+                    s, options, out, ConverterTime64WithMultiplier(
+                        s.primitive_column, s.datetime_multiplier));
+            break;
+
         default:
             throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected column type: {}", s.primitive_column->getFamilyName());
     }

src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h

@@ -41,6 +41,7 @@ DEFINE_ICEBERG_FIELD(time);
 DEFINE_ICEBERG_FIELD(timestamp);
 DEFINE_ICEBERG_FIELD(timestamptz);
 DEFINE_ICEBERG_FIELD(type)
+DEFINE_ICEBERG_FIELD(logicalType); /// this field has a camelCase name
 DEFINE_ICEBERG_FIELD(transform);
 DEFINE_ICEBERG_FIELD(direction);
 

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp

@@ -12,6 +12,7 @@
 #include <Core/TypeId.h>
 #include <DataTypes/DataTypeNullable.h>
 #include <DataTypes/DataTypeString.h>
+#include <DataTypes/DataTypeTime64.h>
 #include <DataTypes/DataTypesNumber.h>
 #include <DataTypes/IDataType.h>
 #include <Databases/DataLake/Common.h>
@@ -129,6 +130,8 @@ bool canDumpIcebergStats(const Field & field, DataTypePtr type)
         case TypeIndex::Date32:
         case TypeIndex::Int64:
         case TypeIndex::DateTime64:
+        case TypeIndex::Time:
+        case TypeIndex::Time64:
         case TypeIndex::String:
             return true;
         default:
@@ -144,6 +147,46 @@ std::vector<uint8_t> dumpValue(T value)
     return bytes;
 }
 
+DataTypePtr getTimeTypeOrNull(DataTypePtr type)
+{
+    if (type->isNullable())
+        return getTimeTypeOrNull(assert_cast<const DataTypeNullable *>(type.get())->getNestedType());
+
+    const WhichDataType which(type);
+    if (which.isTime() || which.isTime64())
+        return type;
+
+    return nullptr;
+}
+
+Int64 getTimeValueInMicroseconds(const Field & field, DataTypePtr type)
+{
+    if (type->isNullable())
+        return getTimeValueInMicroseconds(field, assert_cast<const DataTypeNullable *>(type.get())->getNestedType());
+
+    const WhichDataType which(type);
+    if (which.isTime())
+    {
+        if (field.getType() == Field::Types::Int64)
+            return field.safeGet<Int64>() * 1'000'000;
+        if (field.getType() == Field::Types::UInt64)
+            return static_cast<Int64>(field.safeGet<UInt64>()) * 1'000'000;
+        return static_cast<Int64>(field.safeGet<Int32>()) * 1'000'000;
+    }
+
+    if (which.isTime64())
+    {
+        const auto scale = getDecimalScale(*type);
+        if (scale > 6)
+            throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unsupported type for iceberg {}", type->getName());
+
+        const auto value = field.safeGet<Decimal64>().getValue().value;
+        return value * DataTypeTime64::getScaleMultiplier(6 - scale).value;
+    }
+
+    throw Exception(ErrorCodes::LOGICAL_ERROR, "Expected Time or Time64, got {}", type->getName());
+}
+
 std::vector<uint8_t> dumpFieldToBytes(const Field & field, DataTypePtr type)
 {
     switch (type->getTypeId())
@@ -154,8 +197,12 @@ std::vector<uint8_t> dumpFieldToBytes(const Field & field, DataTypePtr type)
         case TypeIndex::Date:
         case TypeIndex::Date32:
             return dumpValue(field.safeGet<Int32>());
+        case TypeIndex::Time:
+            return dumpValue(getTimeValueInMicroseconds(field, type));
         case TypeIndex::Int64:
             return dumpValue(field.safeGet<Int64>());
+        case TypeIndex::Time64:
+            return dumpValue(getTimeValueInMicroseconds(field, type));
         case TypeIndex::DateTime64:
             return dumpValue(field.safeGet<Decimal64>().getValue().value);
         case TypeIndex::String:
@@ -383,7 +430,16 @@ void extendSchemaForPartitions(
         Poco::JSON::Object::Ptr field = new Poco::JSON::Object;
         field->set(Iceberg::f_field_id, 1000 + i);
         field->set(Iceberg::f_name, partition_columns[i]);
-        field->set(Iceberg::f_type, getAvroType(partition_types[i]));
+        auto logical_type = getAvroLogicalType(partition_types[i]);
+        if (!logical_type.isEmpty())
+        {
+            Poco::JSON::Object::Ptr type_field = new Poco::JSON::Object;
+            type_field->set(Iceberg::f_type, getAvroType(partition_types[i]));
+            type_field->set(Iceberg::f_logicalType, logical_type);
+            field->set(Iceberg::f_type, type_field);
+        }
+        else
+            field->set(Iceberg::f_type, getAvroType(partition_types[i]));
         partition_fields->add(field);
     }
 
@@ -602,6 +658,14 @@ void generateManifestFile(
         avro::GenericRecord & partition_record = data_file.field("partition").value<avro::GenericRecord>();
         for (size_t i = 0; i < partition_columns.size(); ++i)
         {
+            auto partition_time_type = getTimeTypeOrNull(partition_types[i]);
+            if (!partition_values[i].isNull() && partition_time_type)
+            {
+                partition_record.field(partition_columns[i]) =
+                    avro::GenericDatum(getTimeValueInMicroseconds(partition_values[i], partition_types[i]));
+                continue;
+            }
+
             switch (partition_values[i].getType())
             {
                 case Field::Types::Int64:
@@ -629,7 +693,6 @@ void generateManifestFile(
                     partition_record.field(partition_columns[i]) =
                         avro::GenericDatum(partition_values[i].safeGet<Decimal64>().getValue());
                     break;
-
                 case Field::Types::Null:
                     break;
 

src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp

@@ -243,7 +243,7 @@ DataTypePtr IcebergSchemaProcessor::getSimpleType(const String & type_name, Cont
     if (type_name == f_date)
         return std::make_shared<DataTypeDate32>();
     if (type_name == f_time)
-        return std::make_shared<DataTypeInt64>();
+        return std::make_shared<DataTypeTime64>(6);
     if (type_name == f_timestamp)
         return std::make_shared<DataTypeDateTime64>(6);
     if (type_name == f_timestamptz)