timezone_for_iceberg_timestamptz setting

ianton-ru · ianton-ru · commit 58eb474b95df · 2025-10-24T16:26:17.000+02:00
diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp
@@ -6873,6 +6873,16 @@ Use roaring bitmap for iceberg positional deletes.
 )", 0) \
     DECLARE(Bool, export_merge_tree_part_overwrite_file_if_exists, false, R"(
 Overwrite file if it already exists when exporting a merge tree part
+)", 0) \
+    DECLARE(Timezone, timezone_for_iceberg_timestamptz, "UTC", R"(
+Timezone for Iceberg timestamptz field.
+
+Possible values:
+
+- Any valid timezone, e.g. `Europe/Berlin`, `UTC` or `Zulu`
+- `` (empty value) - use session timezone
+
+Default value is `UTC`.
 )", 0) \
     \
     /* ####################################################### */ \
diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp
@@ -47,6 +47,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"allow_retries_in_cluster_requests", false, false, "New setting"},
             {"object_storage_remote_initiator", false, false, "New setting."},
             {"allow_experimental_export_merge_tree_part", false, true, "Turned ON by default for Antalya."},
+            {"timezone_for_iceberg_timestamptz", "UTC", "UTC", "New setting."}
         });
         addSettingsChanges(settings_changes_history, "25.8",
         {
diff --git a/src/Databases/DataLake/Common.cpp b/src/Databases/DataLake/Common.cpp
@@ -61,14 +61,14 @@ std::vector<String> splitTypeArguments(const String & type_str)
     return args;
 }
 
-DB::DataTypePtr getType(const String & type_name, bool nullable, const String & prefix)
+DB::DataTypePtr getType(const String & type_name, bool nullable, DB::ContextPtr context, const String & prefix)
 {
     String name = trim(type_name);
 
     if (name.starts_with("array<") && name.ends_with(">"))
     {
         String inner = name.substr(6, name.size() - 7);
-        return std::make_shared<DB::DataTypeArray>(getType(inner, nullable));
+        return std::make_shared<DB::DataTypeArray>(getType(inner, nullable, context));
     }
 
     if (name.starts_with("map<") && name.ends_with(">"))
@@ -79,7 +79,7 @@ DB::DataTypePtr getType(const String & type_name, bool nullable, const String &
         if (args.size() != 2)
             throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Invalid data type {}", type_name);
 
-        return std::make_shared<DB::DataTypeMap>(getType(args[0], false), getType(args[1], nullable));
+        return std::make_shared<DB::DataTypeMap>(getType(args[0], false, context), getType(args[1], nullable, context));
     }
 
     if (name.starts_with("struct<") && name.ends_with(">"))
@@ -101,13 +101,13 @@ DB::DataTypePtr getType(const String & type_name, bool nullable, const String &
             String full_field_name = prefix.empty() ? field_name : prefix + "." + field_name;
 
             field_names.push_back(full_field_name);
-            field_types.push_back(getType(field_type, nullable, full_field_name));
+            field_types.push_back(getType(field_type, nullable, context, full_field_name));
         }
         return std::make_shared<DB::DataTypeTuple>(field_types, field_names);
     }
 
-    return nullable ? DB::makeNullable(DB::Iceberg::IcebergSchemaProcessor::getSimpleType(name))
-                    : DB::Iceberg::IcebergSchemaProcessor::getSimpleType(name);
+    return nullable ? DB::makeNullable(DB::Iceberg::IcebergSchemaProcessor::getSimpleType(name, context))
+                    : DB::Iceberg::IcebergSchemaProcessor::getSimpleType(name, context);
 }
 
 std::pair<std::string, std::string> parseTableName(const std::string & name)
diff --git a/src/Databases/DataLake/Common.h b/src/Databases/DataLake/Common.h
@@ -2,6 +2,7 @@
 
 #include <Core/NamesAndTypes.h>
 #include <Core/Types.h>
+#include <Interpreters/Context_fwd.h>
 
 namespace DataLake
 {
@@ -10,7 +11,7 @@ String trim(const String & str);
 
 std::vector<String> splitTypeArguments(const String & type_str);
 
-DB::DataTypePtr getType(const String & type_name, bool nullable, const String & prefix = "");
+DB::DataTypePtr getType(const String & type_name, bool nullable, DB::ContextPtr context, const String & prefix = "");
 
 /// Parse a string, containing at least one dot, into a two substrings:
 /// A.B.C.D.E -> A.B.C.D and E, where
diff --git a/src/Databases/DataLake/GlueCatalog.cpp b/src/Databases/DataLake/GlueCatalog.cpp
@@ -372,7 +372,7 @@ bool GlueCatalog::tryGetTableMetadata(
                         column_type = "timestamptz";
                 }
 
-                schema.push_back({column.GetName(), getType(column_type, can_be_nullable)});
+                schema.push_back({column.GetName(), getType(column_type, can_be_nullable, getContext())});
             }
             result.setSchema(schema);
         }
diff --git a/src/Databases/DataLake/HiveCatalog.cpp b/src/Databases/DataLake/HiveCatalog.cpp
@@ -130,7 +130,7 @@ bool HiveCatalog::tryGetTableMetadata(const std::string & namespace_name, const
         auto columns = table.sd.cols;
         for (const auto & column : columns)
         {
-            schema.push_back({column.name, getType(column.type, true)});
+            schema.push_back({column.name, getType(column.type, true, getContext())});
         }
         result.setSchema(schema);
     }
diff --git a/src/Databases/DataLake/RestCatalog.cpp b/src/Databases/DataLake/RestCatalog.cpp
@@ -623,7 +623,7 @@ bool RestCatalog::getTableMetadataImpl(
     if (result.requiresSchema())
     {
         // int format_version = metadata_object->getValue<int>("format-version");
-        auto schema_processor = DB::Iceberg::IcebergSchemaProcessor();
+        auto schema_processor = DB::Iceberg::IcebergSchemaProcessor(getContext());
         auto id = DB::IcebergMetadata::parseTableSchema(metadata_object, schema_processor, log);
         auto schema = schema_processor.getClickhouseTableSchemaById(id);
         result.setSchema(*schema);
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp
@@ -135,7 +135,7 @@ IcebergMetadata::IcebergMetadata(
     : object_storage(std::move(object_storage_))
     , configuration(std::move(configuration_))
     , persistent_components(PersistentTableComponents{
-          .schema_processor = std::make_shared<IcebergSchemaProcessor>(),
+          .schema_processor = std::make_shared<IcebergSchemaProcessor>(context_),
           .metadata_cache = cache_ptr,
           .format_version = format_version_,
           .table_location = metadata_object_->getValue<String>(f_location)
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.cpp
@@ -17,6 +17,7 @@
 #include <base/types.h>
 #include <Common/SharedLockGuard.h>
 #include <base/scope_guard.h>
+#include <Core/Settings.h>
 
 #include <DataTypes/DataTypeArray.h>
 #include <DataTypes/DataTypeDate.h>
@@ -32,6 +33,8 @@
 #include <DataTypes/NestedUtils.h>
 #include <DataTypes/DataTypesNumber.h>
 #include <Formats/FormatFactory.h>
+#include <Interpreters/Context_fwd.h>
+#include <Interpreters/Context.h>
 
 #include <IO/ReadHelpers.h>
 
@@ -46,6 +49,10 @@ extern const int LOGICAL_ERROR;
 extern const int BAD_ARGUMENTS;
 }
 
+namespace Setting
+{
+extern const SettingsTimezone timezone_for_iceberg_timestamptz;
+}
 
 namespace
 {
@@ -221,7 +228,7 @@ NamesAndTypesList IcebergSchemaProcessor::tryGetFieldsCharacteristics(Int32 sche
     return fields;
 }
 
-DataTypePtr IcebergSchemaProcessor::getSimpleType(const String & type_name)
+DataTypePtr IcebergSchemaProcessor::getSimpleType(const String & type_name, ContextPtr context_)
 {
     if (type_name == f_boolean)
         return DataTypeFactory::instance().get("Bool");
@@ -240,7 +247,10 @@ DataTypePtr IcebergSchemaProcessor::getSimpleType(const String & type_name)
     if (type_name == f_timestamp)
         return std::make_shared<DataTypeDateTime64>(6);
     if (type_name == f_timestamptz)
-        return std::make_shared<DataTypeDateTime64>(6, "UTC");
+    {
+        std::string timezone = context_->getSettingsRef()[Setting::timezone_for_iceberg_timestamptz];
+        return std::make_shared<DataTypeDateTime64>(6, timezone);
+    }
     if (type_name == f_string || type_name == f_binary)
         return std::make_shared<DataTypeString>();
     if (type_name == f_uuid)
@@ -323,7 +333,11 @@ IcebergSchemaProcessor::getComplexTypeFromObject(const Poco::JSON::Object::Ptr &
 }
 
 DataTypePtr IcebergSchemaProcessor::getFieldType(
-    const Poco::JSON::Object::Ptr & field, const String & type_key, bool required, String & current_full_name, bool is_subfield_of_root)
+    const Poco::JSON::Object::Ptr & field,
+    const String & type_key,
+    bool required,
+    String & current_full_name,
+    bool is_subfield_of_root)
 {
     if (field->isObject(type_key))
         return getComplexTypeFromObject(field->getObject(type_key), current_full_name, is_subfield_of_root);
@@ -332,7 +346,7 @@ DataTypePtr IcebergSchemaProcessor::getFieldType(
     if (type.isString())
     {
         const String & type_name = type.extract<String>();
-        auto data_type = getSimpleType(type_name);
+        auto data_type = getSimpleType(type_name, getContext());
         return required ? data_type : makeNullable(data_type);
     }
 
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.h
@@ -75,13 +75,15 @@ ColumnMapperPtr createColumnMapper(Poco::JSON::Object::Ptr schema_object);
  *     }
  * }
  */
-class IcebergSchemaProcessor
+class IcebergSchemaProcessor : private WithContext
 {
     static std::string default_link;
 
     using Node = ActionsDAG::Node;
 
 public:
+    IcebergSchemaProcessor(ContextPtr context_) : WithContext(context_) {}
+
     void addIcebergTableSchema(Poco::JSON::Object::Ptr schema_ptr);
     std::shared_ptr<NamesAndTypesList> getClickhouseTableSchemaById(Int32 id);
     std::shared_ptr<const ActionsDAG> getSchemaTransformationDagByIds(Int32 old_id, Int32 new_id);
@@ -92,7 +94,7 @@ class IcebergSchemaProcessor
     Poco::JSON::Object::Ptr getIcebergTableSchemaById(Int32 id) const;
     bool hasClickhouseTableSchemaById(Int32 id) const;
 
-    static DataTypePtr getSimpleType(const String & type_name);
+    static DataTypePtr getSimpleType(const String & type_name, ContextPtr context_);
 
     static std::unordered_map<String, Int64> traverseSchema(Poco::JSON::Array::Ptr schema);
 

Original file line number	Diff line number	Diff line change
`@@ -47,6 +47,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()`
`47`	`47`	`{"allow_retries_in_cluster_requests", false, false, "New setting"},`
`48`	`48`	`{"object_storage_remote_initiator", false, false, "New setting."},`
`49`	`49`	`{"allow_experimental_export_merge_tree_part", false, true, "Turned ON by default for Antalya."},`
	`50`	`+ {"timezone_for_iceberg_timestamptz", "UTC", "UTC", "New setting."}`
`50`	`51`	`});`
`51`	`52`	`addSettingsChanges(settings_changes_history, "25.8",`
`52`	`53`	`{`
Original file line number	Diff line number	Diff line change
`@@ -61,14 +61,14 @@ std::vector<String> splitTypeArguments(const String & type_str)`
`61`	`61`	`return args;`
`62`	`62`	`}`
`63`	`63`
`64`		`-DB::DataTypePtr getType(const String & type_name, bool nullable, const String & prefix)`
	`64`	`+DB::DataTypePtr getType(const String & type_name, bool nullable, DB::ContextPtr context, const String & prefix)`
`65`	`65`	`{`
`66`	`66`	`String name = trim(type_name);`
`67`	`67`
`68`	`68`	`if (name.starts_with("array<") && name.ends_with(">"))`
`69`	`69`	`{`
`70`	`70`	`String inner = name.substr(6, name.size() - 7);`
`71`		`- return std::make_shared<DB::DataTypeArray>(getType(inner, nullable));`
	`71`	`+ return std::make_shared<DB::DataTypeArray>(getType(inner, nullable, context));`
`72`	`72`	`}`
`73`	`73`
`74`	`74`	`if (name.starts_with("map<") && name.ends_with(">"))`
`@@ -79,7 +79,7 @@ DB::DataTypePtr getType(const String & type_name, bool nullable, const String &`
`79`	`79`	`if (args.size() != 2)`
`80`	`80`	`throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Invalid data type {}", type_name);`
`81`	`81`
`82`		`- return std::make_shared<DB::DataTypeMap>(getType(args[0], false), getType(args[1], nullable));`
	`82`	`+ return std::make_shared<DB::DataTypeMap>(getType(args[0], false, context), getType(args[1], nullable, context));`
`83`	`83`	`}`
`84`	`84`
`85`	`85`	`if (name.starts_with("struct<") && name.ends_with(">"))`
`@@ -101,13 +101,13 @@ DB::DataTypePtr getType(const String & type_name, bool nullable, const String &`
`101`	`101`	`String full_field_name = prefix.empty() ? field_name : prefix + "." + field_name;`
`102`	`102`
`103`	`103`	`field_names.push_back(full_field_name);`
`104`		`- field_types.push_back(getType(field_type, nullable, full_field_name));`
	`104`	`+ field_types.push_back(getType(field_type, nullable, context, full_field_name));`
`105`	`105`	`}`
`106`	`106`	`return std::make_shared<DB::DataTypeTuple>(field_types, field_names);`
`107`	`107`	`}`
`108`	`108`
`109`		`- return nullable ? DB::makeNullable(DB::Iceberg::IcebergSchemaProcessor::getSimpleType(name))`
`110`		`- : DB::Iceberg::IcebergSchemaProcessor::getSimpleType(name);`
	`109`	`+ return nullable ? DB::makeNullable(DB::Iceberg::IcebergSchemaProcessor::getSimpleType(name, context))`
	`110`	`+ : DB::Iceberg::IcebergSchemaProcessor::getSimpleType(name, context);`
`111`	`111`	`}`
`112`	`112`
`113`	`113`	`std::pair<std::string, std::string> parseTableName(const std::string & name)`
Original file line number	Diff line number	Diff line change
`@@ -372,7 +372,7 @@ bool GlueCatalog::tryGetTableMetadata(`
`372`	`372`	`column_type = "timestamptz";`
`373`	`373`	`}`
`374`	`374`
`375`		`- schema.push_back({column.GetName(), getType(column_type, can_be_nullable)});`
	`375`	`+ schema.push_back({column.GetName(), getType(column_type, can_be_nullable, getContext())});`
`376`	`376`	`}`
`377`	`377`	`result.setSchema(schema);`
`378`	`378`	`}`
Original file line number	Diff line number	Diff line change
`@@ -130,7 +130,7 @@ bool HiveCatalog::tryGetTableMetadata(const std::string & namespace_name, const`
`130`	`130`	`auto columns = table.sd.cols;`
`131`	`131`	`for (const auto & column : columns)`
`132`	`132`	`{`
`133`		`- schema.push_back({column.name, getType(column.type, true)});`
	`133`	`+ schema.push_back({column.name, getType(column.type, true, getContext())});`
`134`	`134`	`}`
`135`	`135`	`result.setSchema(schema);`
`136`	`136`	`}`
Original file line number	Diff line number	Diff line change
`@@ -623,7 +623,7 @@ bool RestCatalog::getTableMetadataImpl(`
`623`	`623`	`if (result.requiresSchema())`
`624`	`624`	`{`
`625`	`625`	`// int format_version = metadata_object->getValue<int>("format-version");`
`626`		`- auto schema_processor = DB::Iceberg::IcebergSchemaProcessor();`
	`626`	`+ auto schema_processor = DB::Iceberg::IcebergSchemaProcessor(getContext());`
`627`	`627`	`auto id = DB::IcebergMetadata::parseTableSchema(metadata_object, schema_processor, log);`
`628`	`628`	`auto schema = schema_processor.getClickhouseTableSchemaById(id);`
`629`	`629`	`result.setSchema(*schema);`