Skip to content

Antalya 26.1: Support for 'time' type in Iceberg, reading only #1613

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ianton-ru wants to merge 4 commits into
base: antalya-26.1
Choose a base branch
from
Open

Antalya 26.1: Support for 'time' type in Iceberg, reading only #1613

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9f00f37
Iceberg type 'time' support
ianton-ru Mar 18, 2026
bb896c3
Fix tests
ianton-ru Mar 19, 2026
28ba40b
Fix
ianton-ru Mar 19, 2026
f0e1038
Fix test
ianton-ru Apr 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion src/Processors/Formats/Impl/AvroRowInputFormat.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,6 +156,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;
Expand Down Expand Up @@ -304,6 +307,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())
Expand Down Expand Up @@ -1283,8 +1288,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>()};
}
Expand All @@ -1295,6 +1303,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>();
}
Expand Down
4 changes: 4 additions & 0 deletions src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,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:
Expand All @@ -155,7 +157,9 @@ std::vector<uint8_t> dumpFieldToBytes(const Field & field, DataTypePtr type)
case TypeIndex::Date32:
return dumpValue(field.safeGet<Int32>());
case TypeIndex::Int64:
case TypeIndex::Time:
return dumpValue(field.safeGet<Int64>());
case TypeIndex::Time64:
case TypeIndex::DateTime64:
return dumpValue(field.safeGet<Decimal64>().getValue().value);
case TypeIndex::String:
Expand Down
2 changes: 1 addition & 1 deletion src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Comment on lines 245 to +246
Copy link
Copy Markdown

@chatgpt-codex-connector chatgpt-codex-connector bot Apr 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Support Time64 in manifest Avro type mapping

After f_time is remapped to DataTypeTime64(6), inserts into Iceberg tables partitioned by a time column can fail in the write path: ChunkPartitioner::getResultTypes() returns Time64, generateManifestFile() passes that into extendSchemaForPartitions(), and getAvroType() still has no TypeIndex::Time64 branch (it only handles TypeIndex::Time), so it throws Unsupported type for iceberg Time64(...). This is a regression from the previous Int64 mapping because writes for time-partitioned tables now error during manifest generation.

Useful? React with 👍 / 👎.

if (type_name == f_timestamp)
return std::make_shared<DataTypeDateTime64>(6);
if (type_name == f_timestamptz)
Expand Down
1 change: 1 addition & 0 deletions src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -402,6 +402,7 @@ std::pair<Poco::Dynamic::Var, bool> getIcebergType(DataTypePtr type, Int32 & ite
case TypeIndex::DateTime64:
return {"timestamp", true};
case TypeIndex::Time:
case TypeIndex::Time64:
return {"time", true};
case TypeIndex::String:
return {"string", true};
Expand Down
97 changes: 95 additions & 2 deletions tests/integration/test_database_iceberg/test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import random
import time
import uuid
from datetime import datetime, timedelta
from datetime import datetime, timedelta, time as dtime

import pyarrow as pa
import pytest
Expand All @@ -26,7 +26,8 @@
StringType,
StructType,
TimestampType,
TimestamptzType
TimestamptzType,
TimeType,
)
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER

Expand Down Expand Up @@ -1015,3 +1016,95 @@ def _test_cluster_joins(started_cluster):
)

assert res == "Jack\tBlack\nJack\tSilver\nJohn\tBlack\nJohn\tSilver\n"


@pytest.mark.parametrize("storage_type", ["s3"])
def test_partitioning_by_time(started_cluster, storage_type):
node = started_cluster.instances["node1"]

test_ref = f"test_partitioning_by_time_{uuid.uuid4()}"
table_name = f"{test_ref}_table"
root_namespace = f"{test_ref}_namespace"

namespace = f"{root_namespace}.A"
catalog = load_catalog_impl(started_cluster)
catalog.create_namespace(namespace)

schema = Schema(
NestedField(
field_id=1,
name="key",
field_type=TimeType(),
required=False
),
NestedField(
field_id=2,
name="value",
field_type=StringType(),
required=False,
),
)

partition_spec = PartitionSpec(
PartitionField(
source_id=1, field_id=1000, transform=IdentityTransform(), name="partition_key"
)
)

table = create_table(catalog, namespace, table_name, schema=schema, partition_spec=partition_spec)
data = [{"key": dtime(12,0,0), "value": "test"}]
df = pa.Table.from_pylist(data)
table.append(df)

create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME)

assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}`") == "12:00:00.000000\ttest\n"


@pytest.mark.parametrize("storage_type", ["s3"])
def test_partitioning_by_string(started_cluster, storage_type):
node = started_cluster.instances["node1"]

test_ref = f"test_partitioning_by_string_{uuid.uuid4()}"
table_name = f"{test_ref}_table"
root_namespace = f"{test_ref}_namespace"

namespace = f"{root_namespace}.A"
catalog = load_catalog_impl(started_cluster)
catalog.create_namespace(namespace)

schema = Schema(
NestedField(
field_id=1,
name="key",
field_type=StringType(),
required=False
),
NestedField(
field_id=2,
name="value",
field_type=StringType(),
required=False,
),
NestedField(
field_id=3,
name="time_value",
field_type=TimeType(),
required=False,
),
)

partition_spec = PartitionSpec(
PartitionField(
source_id=1, field_id=1000, transform=IdentityTransform(), name="partition_key"
)
)

table = create_table(catalog, namespace, table_name, schema=schema, partition_spec=partition_spec)
data = [{"key": "a:b,c[d=e/f%g?h", "value": "test", "time_value": dtime(12,0,0)}]
df = pa.Table.from_pylist(data)
table.append(df)

create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME)

assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}`") == "a:b,c[d=e/f%g?h\ttest\t12:00:00.000000\n"
Loading