fix: allow avro date reads from plain int (#728)

per https://github.com/apache/iceberg/pull/16446/changes

Author: zhjwpku

src/iceberg/avro/avro_data_util.cc

@@ -395,10 +395,10 @@ Status AppendPrimitiveValueToBuilder(const ::avro::NodePtr& avro_node,
     }
 
     case TypeId::kDate: {
-      if (avro_node->type() != ::avro::AVRO_INT ||
-          avro_node->logicalType().type() != ::avro::LogicalType::DATE) {
+      if (!IsAvroDateOrPlainInt(avro_node)) {
         return InvalidArgument(
-            "Expected Avro int with DATE logical type for date field, got: {}",
+            "Expected Avro int with DATE logical type or plain int for date field, got: "
+            "{}",
             ToString(avro_node));
       }
       auto* builder = internal::checked_cast<::arrow::Date32Builder*>(array_builder);

src/iceberg/avro/avro_direct_decoder.cc

@@ -523,10 +523,10 @@ Status DecodePrimitiveValueToBuilder(const ::avro::NodePtr& avro_node,
     }
 
     case TypeId::kDate: {
-      if (avro_node->type() != ::avro::AVRO_INT ||
-          avro_node->logicalType().type() != ::avro::LogicalType::DATE) {
+      if (!IsAvroDateOrPlainInt(avro_node)) {
         return InvalidArgument(
-            "Expected Avro int with DATE logical type for date field, got: {}",
+            "Expected Avro int with DATE logical type or plain int for date field, got: "
+            "{}",
             ToString(avro_node));
       }
       auto* builder = internal::checked_cast<::arrow::Date32Builder*>(array_builder);

src/iceberg/avro/avro_schema_util.cc

@@ -125,6 +125,12 @@ std::string ToString(const ::avro::LogicalType::Type& logical_type) {
   return ToString(::avro::LogicalType(logical_type));
 }
 
+bool IsAvroDateOrPlainInt(const ::avro::NodePtr& node) {
+  return node->type() == ::avro::AVRO_INT &&
+         (node->logicalType().type() == ::avro::LogicalType::DATE ||
+          node->logicalType().type() == ::avro::LogicalType::NONE);
+}
+
 Status ToAvroNodeVisitor::Visit(const BooleanType& type, ::avro::NodePtr* node) {
   *node = std::make_shared<::avro::NodePrimitive>(::avro::AVRO_BOOL);
   return {};
@@ -550,8 +556,7 @@ Status ValidateAvroSchemaEvolution(const Type& expected_type,
       }
       break;
     case TypeId::kDate:
-      if (avro_node->type() == ::avro::AVRO_INT &&
-          HasLogicalType(avro_node, ::avro::LogicalType::DATE)) {
+      if (IsAvroDateOrPlainInt(avro_node)) {
         return {};
       }
       break;

src/iceberg/avro/avro_schema_util_internal.h

@@ -147,6 +147,14 @@ std::string ToString(const ::avro::NodePtr& node);
 std::string ToString(const ::avro::LogicalType& logical_type);
 std::string ToString(const ::avro::LogicalType::Type& logical_type);
 
+/// \brief Check if an Avro node can be read as an Iceberg date.
+///
+/// Iceberg dates are encoded as Avro ints with the date logical type. Readers
+/// also accept plain ints where the expected Iceberg type is date.
+/// \param node The Avro node to check.
+/// \return True if the node is an Avro date or a plain Avro int.
+bool IsAvroDateOrPlainInt(const ::avro::NodePtr& node);
+
 /// \brief Check if an Avro node has a map logical type.
 /// \param node The Avro node to check.
 /// \return True if the node has a map logical type, false otherwise.

src/iceberg/test/avro_schema_test.cc

@@ -950,6 +950,30 @@ TEST(AvroSchemaProjectionTest, ProjectSchemaEvolutionIntToLong) {
   ASSERT_EQ(std::get<1>(projection.fields[0].from), 0);
 }
 
+TEST(AvroSchemaProjectionTest, ProjectDateFromPlainInt) {
+  Schema expected_schema({
+      SchemaField::MakeRequired(/*field_id=*/1, "day", iceberg::date()),
+  });
+
+  std::string avro_schema_json = R"({
+    "type": "record",
+    "name": "iceberg_schema",
+    "fields": [
+      {"name": "day", "type": "int", "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, IsOk());
+
+  const auto& projection = *projection_result;
+  ASSERT_EQ(projection.fields.size(), 1);
+  ASSERT_EQ(projection.fields[0].kind, FieldProjection::Kind::kProjected);
+  ASSERT_EQ(std::get<1>(projection.fields[0].from), 0);
+}
+
 TEST(AvroSchemaProjectionTest, ProjectSchemaEvolutionFloatToDouble) {
   // Create iceberg schema expecting a double
   Schema expected_schema({