diff --git a/cmake_modules/IcebergThirdpartyToolchain.cmake b/cmake_modules/IcebergThirdpartyToolchain.cmake index c93e914ae..c38d96028 100644 --- a/cmake_modules/IcebergThirdpartyToolchain.cmake +++ b/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 diff --git a/src/iceberg/avro/avro_schema_util.cc b/src/iceberg/avro/avro_schema_util.cc index 5b10fd05e..94d0a6116 100644 --- a/src/iceberg/avro/avro_schema_util.cc +++ b/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 GetAdjustToUtc(const ::avro::NodePtr& node) { if (node->customAttributes() == 0) { return std::nullopt; diff --git a/src/iceberg/test/avro_schema_test.cc b/src/iceberg/test/avro_schema_test.cc index a65a0abdd..f147a607b 100644 --- a/src/iceberg/test/avro_schema_test.cc +++ b/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(); diff --git a/src/iceberg/test/avro_test.cc b/src/iceberg/test/avro_test.cc index d322c3a8f..aaee666df 100644 --- a/src/iceberg/test/avro_test.cc +++ b/src/iceberg/test/avro_test.cc @@ -24,6 +24,8 @@ #include #include #include +#include +#include #include #include @@ -489,6 +491,19 @@ TEST_P(AvroReaderParameterizedTest, DateTimeTypes) { WriteAndVerify(schema, expected_string); } +TEST_P(AvroReaderParameterizedTest, TimestampTzTypes) { + auto schema = std::make_shared(std::vector{ + 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(std::vector{ SchemaField::MakeRequired(1, "id", std::make_shared()), @@ -1202,6 +1217,41 @@ TEST_P(AvroWriterTest, WriteTemporalTypes) { VerifyWrittenData(test_data); } +TEST_P(AvroWriterTest, WriteTimestampTzTypes) { + auto schema = std::make_shared(std::vector{ + 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> 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(std::vector{ SchemaField::MakeRequired(1, "id", std::make_shared()),