chore: bump avro to 209a373 to preserve adjust-to-utc on primitives

cmake_modules/IcebergThirdpartyToolchain.cmake

@@ -314,7 +314,7 @@ function(resolve_avro_dependency)
     fetchcontent_declare(avro-cpp
                          ${FC_DECLARE_COMMON_OPTIONS}
                          GIT_REPOSITORY ${AVRO_GIT_REPOSITORY}
-                         GIT_TAG 997d50d312613e921598aaed30b082f9bcf9c6ea
+                         GIT_TAG 209a3735ec330679790824da54b7558db55e4e7f
                          SOURCE_SUBDIR
                          lang/c++
                          FIND_PACKAGE_ARGS

src/iceberg/avro/avro_schema_util.cc

@@ -493,6 +493,9 @@ bool HasLogicalType(const ::avro::NodePtr& node,
   return node->logicalType().type() == expected_type;
 }
 
+// Relies on Avro preserving custom attributes on primitive nodes (AVRO-4351);
+// before that, "adjust-to-utc" was dropped when parsing a file schema and every
+// timestamp read from a file looked like it had no timezone.
 std::optional<std::string> GetAdjustToUtc(const ::avro::NodePtr& node) {
   if (node->customAttributes() == 0) {
     return std::nullopt;

src/iceberg/test/avro_schema_test.cc

@@ -1339,6 +1339,64 @@ TEST(AvroSchemaProjectionTest, RejectTimestampNsFromMicrosType) {
   ASSERT_THAT(projection_result, HasErrorMessage("Cannot read"));
 }
 
+TEST(AvroSchemaProjectionTest, ProjectTimestampTzFromParsedAdjustToUtc) {
+  Schema expected_schema({
+      SchemaField::MakeRequired(/*field_id=*/1, "ts", iceberg::timestamp_tz()),
+      SchemaField::MakeRequired(/*field_id=*/2, "ts_ns", iceberg::timestamptz_ns()),
+  });
+
+  std::string avro_schema_json = R"({
+    "type": "record",
+    "name": "iceberg_schema",
+    "fields": [
+      {"name": "ts", "type": {
+        "type": "long",
+        "logicalType": "timestamp-micros",
+        "adjust-to-utc": true
+      }, "field-id": 1},
+      {"name": "ts_ns", "type": {
+        "type": "long",
+        "logicalType": "timestamp-nanos",
+        "adjust-to-utc": true
+      }, "field-id": 2}
+    ]
+  })";
+  auto avro_schema = ::avro::compileJsonSchemaFromString(avro_schema_json);
+
+  auto projection_result =
+      Project(expected_schema, avro_schema.root(), /*prune_source=*/false);
+  ASSERT_THAT(projection_result, IsOk());
+
+  const auto& projection = *projection_result;
+  ASSERT_EQ(projection.fields.size(), 2);
+  EXPECT_EQ(projection.fields[0].kind, FieldProjection::Kind::kProjected);
+  EXPECT_EQ(projection.fields[1].kind, FieldProjection::Kind::kProjected);
+}
+
+TEST(AvroSchemaProjectionTest, RejectTimestampFromParsedAdjustToUtc) {
+  Schema expected_schema({
+      SchemaField::MakeRequired(/*field_id=*/1, "ts", iceberg::timestamp()),
+  });
+
+  std::string avro_schema_json = R"({
+    "type": "record",
+    "name": "iceberg_schema",
+    "fields": [
+      {"name": "ts", "type": {
+        "type": "long",
+        "logicalType": "timestamp-micros",
+        "adjust-to-utc": true
+      }, "field-id": 1}
+    ]
+  })";
+  auto avro_schema = ::avro::compileJsonSchemaFromString(avro_schema_json);
+
+  auto projection_result =
+      Project(expected_schema, avro_schema.root(), /*prune_source=*/false);
+  ASSERT_THAT(projection_result, IsError(ErrorKind::kInvalidSchema));
+  ASSERT_THAT(projection_result, HasErrorMessage("Cannot read"));
+}
+
 TEST(AvroSchemaProjectionTest, ProjectMapTypeWithNonStringKey) {
   ::iceberg::avro::RegisterLogicalTypes();
 

src/iceberg/test/avro_test.cc

@@ -24,6 +24,8 @@
 #include <string>
 #include <string_view>
 #include <unordered_map>
+#include <utility>
+#include <vector>
 
 #include <arrow/array.h>
 #include <arrow/array/array_base.h>
@@ -489,6 +491,19 @@ TEST_P(AvroReaderParameterizedTest, DateTimeTypes) {
   WriteAndVerify(schema, expected_string);
 }
 
+TEST_P(AvroReaderParameterizedTest, TimestampTzTypes) {
+  auto schema = std::make_shared<iceberg::Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "timestamptz_col", iceberg::timestamp_tz()),
+      SchemaField::MakeRequired(2, "timestamptz_ns_col", iceberg::timestamptz_ns())});
+
+  std::string expected_string = R"([
+    [1640995200000000, 1640995200000000001],
+    [1641081599000000, 1641081599000000002]
+  ])";
+
+  WriteAndVerify(schema, expected_string);
+}
+
 TEST_P(AvroReaderParameterizedTest, NestedStruct) {
   auto schema = std::make_shared<iceberg::Schema>(std::vector<SchemaField>{
       SchemaField::MakeRequired(1, "id", std::make_shared<IntType>()),
@@ -1202,6 +1217,41 @@ TEST_P(AvroWriterTest, WriteTemporalTypes) {
   VerifyWrittenData(test_data);
 }
 
+TEST_P(AvroWriterTest, WriteTimestampTzTypes) {
+  auto schema = std::make_shared<iceberg::Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "timestamp_col", iceberg::timestamp()),
+      SchemaField::MakeRequired(2, "timestamptz_col", iceberg::timestamp_tz()),
+      SchemaField::MakeRequired(3, "timestamp_ns_col", iceberg::timestamp_ns()),
+      SchemaField::MakeRequired(4, "timestamptz_ns_col", iceberg::timestamptz_ns())});
+
+  std::string test_data = R"([
+    [1640995200000000, 1640995200000000, 1640995200000000001, 1640995200000000001],
+    [1641081599000000, 1641081599000000, 1641081599000000002, 1641081599000000002]
+  ])";
+
+  WriteAvroFile(schema, test_data);
+
+  auto root = PhysicalAvroSchema().root();
+  ASSERT_EQ(root->type(), ::avro::AVRO_RECORD);
+  ASSERT_EQ(root->leaves(), 4);
+  const std::vector<std::pair<::avro::LogicalType::Type, std::string>> expected = {
+      {::avro::LogicalType::TIMESTAMP_MICROS, "false"},
+      {::avro::LogicalType::TIMESTAMP_MICROS, "true"},
+      {::avro::LogicalType::TIMESTAMP_NANOS, "false"},
+      {::avro::LogicalType::TIMESTAMP_NANOS, "true"},
+  };
+  for (size_t i = 0; i < expected.size(); ++i) {
+    auto node = root->leafAt(i);
+    EXPECT_EQ(node->type(), ::avro::AVRO_LONG);
+    EXPECT_EQ(node->logicalType().type(), expected[i].first);
+    ASSERT_EQ(node->customAttributes(), 1);
+    EXPECT_EQ(node->customAttributesAt(0).getAttribute(std::string(kAdjustToUtcProp)),
+              expected[i].second);
+  }
+
+  VerifyWrittenData(test_data);
+}
+
 TEST_P(AvroWriterTest, WriteNestedStruct) {
   auto schema = std::make_shared<iceberg::Schema>(std::vector<SchemaField>{
       SchemaField::MakeRequired(1, "id", std::make_shared<IntType>()),