解析GCP临时表Avro Schema时遭遇‘logical datetime不支持’错误

场景 #

通过BigQuery ReadSession从临时表读取数据并指定Avro格式时，只要源表包含datetime类型列，执行Schema.Parse(response.AvroSchema.Schema)就会触发logical datetime is not supported错误。环境为.NET 8 + Avro 1.12库。

相关代码：

ReadSession rd = new ReadSession();
rd.table = "mytemptable";
rd.DataFormat = DataFormat.Avro;
ReadSession response = bigqueryclient.CreateReadSession(parent, rd, 0);
var schema = Schema.Parse(response.AvroSchema.Schema); // 报错行

解决方案 #

方案1：注册自定义逻辑类型解析器 #

Avro .NET库支持注册自定义逻辑类型处理，针对BigQuery返回的datetime-millis逻辑类型，我们可以添加专属解析器：

// 解析Schema前注册自定义逻辑类型
LogicalTypeRegistry.Register<DateTimeLogicalType>("datetime-millis", (schema, type) => 
    new DateTimeLogicalType(schema, type));

// 正常解析Schema
var schema = Schema.Parse(response.AvroSchema.Schema);

如果默认库中没有DateTimeLogicalType，可以自行实现基础逻辑类型类：

public class DateTimeLogicalType : LogicalType
{
    public DateTimeLogicalType(Schema schema, Type type) : base(schema, type) { }

    public override object ConvertToBaseValue(object logicalValue, Schema schema)
    {
        if (logicalValue is DateTime dt)
        {
            var epoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc);
            return (long)(dt.ToUniversalTime() - epoch).TotalMilliseconds;
        }
        throw new InvalidCastException("Expected DateTime");
    }

    public override object ConvertToLogicalValue(object baseValue, Schema schema)
    {
        if (baseValue is long ms)
        {
            var epoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc);
            return epoch.AddMilliseconds(ms).ToLocalTime();
        }
        throw new InvalidCastException("Expected long");
    }

    public override Type GetCSharpType() => typeof(DateTime);
}

方案2：修改ReadSession配置，输出兼容Avro的类型 #

创建ReadSession时，指定将datetime列转换为Avro原生支持的类型（比如TIMESTAMP或STRING）：

var rd = new ReadSession
{
    table = "mytemptable",
    DataFormat = DataFormat.Avro,
    ReadOptions = new ReadSession.Types.ReadOptions
    {
        Schema = new TableSchema
        {
            Fields = 
            {
                // 将datetime列"my_datetime"转换为TIMESTAMP
                new TableFieldSchema
                {
                    Name = "my_datetime",
                    Type = "TIMESTAMP",
                    Mode = "NULLABLE"
                }
                // 其他字段保留原类型
            }
        }
    }
};

ReadSession response = bigqueryclient.CreateReadSession(parent, rd, 0);
var schema = Schema.Parse(response.AvroSchema.Schema);

如果需要批量转换所有datetime字段，可以先获取原表Schema再处理：

// 获取原表Schema
var table = bigqueryclient.GetTable(rd.table);
var convertedSchema = new TableSchema();

foreach (var field in table.Schema.Fields)
{
    if (field.Type == "DATETIME")
    {
        convertedSchema.Fields.Add(new TableFieldSchema
        {
            Name = field.Name,
            Type = "TIMESTAMP",
            Mode = field.Mode
        });
    }
    else
    {
        convertedSchema.Fields.Add(field);
    }
}

rd.ReadOptions = new ReadSession.Types.ReadOptions { Schema = convertedSchema };

方案3：手动修改Avro Schema字符串 #

解析前直接替换Schema中的datetime-millis逻辑类型为Avro支持的类型，比如timestamp-millis：

string avroSchemaStr = response.AvroSchema.Schema;
// 替换datetime-millis为timestamp-millis
avroSchemaStr = avroSchemaStr.Replace("\"logicalType\": \"datetime-millis\"", "\"logicalType\": \"timestamp-millis\"");

var schema = Schema.Parse(avroSchemaStr);

如果Schema结构复杂，建议用JSON解析库批量修改：

using Newtonsoft.Json.Linq;

JObject schemaObj = JObject.Parse(response.AvroSchema.Schema);
// 遍历所有字段修改逻辑类型
foreach (var field in schemaObj["fields"].Children<JObject>())
{
    var typeObj = field["type"] as JObject;
    if (typeObj?["logicalType"]?.Value<string>() == "datetime-millis")
    {
        typeObj["logicalType"] = "timestamp-millis";
    }
}

var modifiedSchemaStr = schemaObj.ToString();
var schema = Schema.Parse(modifiedSchemaStr);

内容的提问来源于stack exchange，提问作者Mohit