AVRO-2825: [csharp] Resolve: C# Logical Types throw exception on unkn…

lang/csharp/src/apache/codegen/Properties/launchSettings.json

@@ -0,0 +1,8 @@
+{
+  "profiles": {
+    "Avro.codegen": {
+      "commandName": "Project",
+      "commandLineArgs": "-s C:\\Users\\Thomas.Bruns\\source\\repos\\AzureDevOps\\TQL.Kafka.Samples\\TQL.Kafka.Samples\\TQL.Kafka.Samples.Messages\\v2\\cdc_customers_dbo_tblcustomers.avsc .\\tab"
+    }
+  }
+}

lang/csharp/src/apache/main/Properties/launchSettings.json

@@ -0,0 +1,8 @@
+{
+  "profiles": {
+    "Avro.main": {
+      "commandName": "Project",
+      "commandLineArgs": "C:\\Users\\Thomas.Bruns\\source\\repos\\AzureDevOps\\TQL.Kafka.Samples\\TQL.Kafka.Samples\\TQL.Kafka.Samples.Messages\\v2\\cdc_customers_dbo_tblcustomers.avsc"
+    }
+  }
+}

lang/csharp/src/apache/main/Schema/LogicalSchema.cs

@@ -54,7 +54,7 @@ private LogicalSchema(Schema baseSchema, string logicalTypeName,  PropertyMap pr
         {
             BaseSchema = baseSchema ?? throw new ArgumentNullException(nameof(baseSchema));
             LogicalTypeName = logicalTypeName;
-            LogicalType = LogicalTypeFactory.Instance.GetFromLogicalSchema(this);
+            LogicalType = LogicalTypeFactory.Instance.GetFromLogicalSchema(this, true);
         }
 
         /// <summary>

lang/csharp/src/apache/main/Util/LogicalTypeFactory.cs

@@ -45,7 +45,7 @@ private LogicalTypeFactory()
                 { TimeMicrosecond.LogicalTypeName, new TimeMicrosecond() },
                 { TimestampMillisecond.LogicalTypeName, new TimestampMillisecond() },
                 { TimestampMicrosecond.LogicalTypeName, new TimestampMicrosecond() },
-                { Uuid.LogicalTypeName, new Uuid() }
+                { Uuid.LogicalTypeName, new Uuid() },
             };
         }
 
@@ -67,22 +67,22 @@ public void Register(LogicalType logicalType)
         /// <returns>A <see cref="LogicalType" />.</returns>
         public LogicalType GetFromLogicalSchema(LogicalSchema schema, bool ignoreInvalidOrUnknown = false)
         {
-            try
-            {
-                if (!_logicalTypes.TryGetValue(schema.LogicalTypeName, out LogicalType logicalType))
-                    throw new AvroTypeException("Logical type '" + schema.LogicalTypeName + "' is not supported.");
+            LogicalType logicalType = null;
 
+            if (_logicalTypes.TryGetValue(schema.LogicalTypeName, out logicalType))
+            {
                 logicalType.ValidateSchema(schema);
-
-                return logicalType;
             }
-            catch (AvroTypeException)
+            else if (ignoreInvalidOrUnknown)
+            {
+                logicalType = new UnknownLogicalType(schema);
+            }
+            else
             {
-                if (!ignoreInvalidOrUnknown)
-                    throw;
+                throw new AvroTypeException("Logical type '" + schema.LogicalTypeName + "' is not supported.");
             }
 
-            return null;
+            return logicalType;
         }
     }
 }

lang/csharp/src/apache/main/Util/UnknownLogicalType.cs

@@ -0,0 +1,57 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace Avro.Util
+{
+    public class UnknownLogicalType : LogicalType
+    {
+        public LogicalSchema Schema { get; }
+
+        public UnknownLogicalType(LogicalSchema schema) : base(schema.LogicalTypeName)
+        {
+            this.Schema = schema;
+        }
+
+        public override object ConvertToBaseValue(object logicalValue, LogicalSchema schema)
+        { 
+            throw new NotImplementedException();
+        }
+
+        public override object ConvertToLogicalValue(object baseValue, LogicalSchema schema)
+        {
+            throw new NotImplementedException();
+        }
+
+        public override Type GetCSharpType(bool nullible)
+        {
+            // handle all Primitive Types
+            switch (this.Schema.BaseSchema.Name)
+            {
+                case @"string":
+                    return typeof(System.String);
+                case @"boolean":
+                    return typeof(System.Boolean);
+                case @"int":
+                    return typeof(System.Int32);
+                case @"long":
+                    return typeof(System.Int64);
+                case @"float":
+                    return typeof(System.Single);
+                case @"double":
+                    return typeof(System.Double);
+                case @"bytes":
+                    return typeof(System.Byte[]);
+                default:
+                    return typeof(System.Object);
+            }
+        }
+
+        public override bool IsInstanceOfLogicalType(object logicalValue)
+        {
+            //         => throw new NotImplementedException();
+            return true;
+        }
+
+    }
+}

lang/csharp/src/apache/test/AvroGen/AvroGenSchemaTests.cs

@@ -619,7 +619,10 @@ public void NotSupportedSchema(string schema, Type expectedException)
                 string schemaFileName = Path.Combine(outputDir, $"{uniqueId}.avsc");
                 System.IO.File.WriteAllText(schemaFileName, schema);
 
-                Assert.That(AvroGenTool.GenSchema(schemaFileName, outputDir, new Dictionary<string, string>(), false), Is.EqualTo(1));
+                // We now support unknown logical types
+                //Assert.That(AvroGenTool.GenSchema(schemaFileName, outputDir, new Dictionary<string, string>(), false), Is.EqualTo(1));
+                Assert.That(AvroGenTool.GenSchema(schemaFileName, outputDir, new Dictionary<string, string>(), false), Is.EqualTo(0));
+
             }
             finally
             {

lang/csharp/src/apache/test/Schema/SchemaTests.cs

@@ -19,6 +19,7 @@
 using System.Collections.Generic;
 using NUnit.Framework;
 using System.Linq;
+using Avro.Util;
 
 namespace Avro.Test
 {
@@ -548,12 +549,77 @@
             testToString(sc);
         }
 
+        // Make sure unknown type is carried thru to LogicalTypeName
         [TestCase("{\"type\": \"int\", \"logicalType\": \"unknown\"}", "unknown")]
         public void TestUnknownLogical(string s, string unknownType)
         {
-            var err = Assert.Throws<AvroTypeException>(() => Schema.Parse(s));
+            //var err = Assert.Throws<AvroTypeException>(() => Schema.Parse(s));
+            //Assert.AreEqual("Logical type '" + unknownType + "' is not supported.", err.Message);
 
-            Assert.AreEqual("Logical type '" + unknownType + "' is not supported.", err.Message);
+            var schema = Schema.Parse(s);
+            Assert.IsInstanceOf(typeof(LogicalSchema), schema);
+
+            var logicalSchema = schema as LogicalSchema;
+            Assert.IsInstanceOf(typeof(UnknownLogicalType), logicalSchema.LogicalType);
+
+            Assert.AreEqual(logicalSchema.LogicalTypeName, unknownType);
+        }
+
+        /*
+            {
+              "fields": [
+                {
+                  "default": 0,
+                  "name": "firstField",
+                  "type": "int"
+                },
+                {
+                  "default": null,
+                  "name": "secondField",
+                  "type": [
+                    "null",
+                    {
+                      "logicalType": "varchar",
+                      "maxLength": 65,
+                      "type": "string"
+                    }
+                  ]
+                }
+              ],
+              "name": "sample_schema",
+              "type": "record"
+            }
+         */
+
+        // Before Change will throw Avro.AvroTypeException: 'Logical type 'varchar' is not supported.'
+        // Per AVRO Spec (v1.8.0 - v1.11.1) ... Logical Types Section
+        //  Language implementations must ignore unknown logical types when reading, and should use the underlying Avro type.
+        [TestCase("{\"fields\": [{\"default\": 0,\"name\": \"firstField\",\"type\": \"int\"},{\"default\": null,\"name\": \"secondField\",\"type\": [\"null\",{\"logicalType\": \"varchar\",\"maxLength\": 65,\"type\": \"string\"}]}],\"name\": \"sample_schema\",\"type\": \"record\"}")]
+        public void TestUnknownLogicalType(string schemaText)
+        {
+            var schema = Avro.Schema.Parse(schemaText);
+            Assert.IsNotNull(schema);
+
+            var secondField = ((RecordSchema)schema).Fields.FirstOrDefault(f => f.Name == @"secondField");
+            Assert.IsNotNull(secondField);
+
+            var secondFieldSchema = ((Field)secondField).Schema;
+            Assert.IsNotNull(secondFieldSchema);
+
+            var secondFieldUnionSchema = (UnionSchema)secondFieldSchema;
+            Assert.IsNotNull(secondFieldUnionSchema);
+
+            var props = secondFieldUnionSchema.Schemas.Where(s => s.Props != null).ToList();
+            Assert.IsNotNull(props);
+            Assert.IsTrue(props.Count == 1);
+
+            var prop = props[0];
+            // Confirm that the unknown logical type is ignored and the underlying AVRO type is used
+            Assert.IsTrue(prop.Name == @"string");
+            var logicalSchema = prop as LogicalSchema;
+            Assert.IsInstanceOf(typeof(UnknownLogicalType), logicalSchema.LogicalType);
+
+            Assert.AreEqual(logicalSchema.LogicalTypeName, @"varchar");
         }
 
         [TestCase("{\"type\": \"map\", \"values\": \"long\"}", "long")]