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

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\\v1\\cdc_tql_dbo_tbldrops.avsc .\\tab --namespace NO_SCHEMA_NAMESPACE:TQL.DEMO"
+    }
+  }
+}

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

@@ -47,14 +47,27 @@ internal static LogicalSchema NewInstance(JToken jtok, PropertyMap props, Schema
             JToken jtype = jtok["type"];
             if (null == jtype) throw new AvroTypeException("Logical Type does not have 'type'");
 
-            return new LogicalSchema(Schema.ParseJson(jtype, names, encspace), JsonHelper.GetRequiredString(jtok, "logicalType"),  props);
+            JToken baseSchemaToken = jtype;
+
+            if (jtok is JObject jo && jtype.Type == JTokenType.String)
+            {
+                string typeStr = (string)jtype;
+                if (typeStr == "record" || typeStr == "enum" || typeStr == "array" || typeStr == "map" || typeStr == "fixed")
+                {
+                    var clone = (JObject)jo.DeepClone();
+                    clone.Property("logicalType")?.Remove();
+                    baseSchemaToken = clone;
+                }
+            }
+
+            return new LogicalSchema(Schema.ParseJson(baseSchemaToken, names, encspace), JsonHelper.GetRequiredString(jtok, "logicalType"), props);
         }
 
         private LogicalSchema(Schema baseSchema, string logicalTypeName,  PropertyMap props) : base(Type.Logical, props)
         {
             BaseSchema = baseSchema ?? throw new ArgumentNullException(nameof(baseSchema));
             LogicalTypeName = logicalTypeName;
-            LogicalType = LogicalTypeFactory.Instance.GetFromLogicalSchema(this);
+            LogicalType = LogicalTypeFactory.Instance.GetFromLogicalSchema(this, true);
         }
 
         /// <summary>
@@ -97,6 +110,7 @@ public override bool CanRead(Schema writerSchema)
             if (writerSchema.Tag != Tag) return false;
 
             LogicalSchema that = writerSchema as LogicalSchema;
+
             return BaseSchema.CanRead(that.BaseSchema);
         }
 

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

@@ -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,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace Avro.Util
+{
+    /// <summary>
+    /// Class UnknownLogicalType.
+    /// Implements the <see cref="Avro.Util.LogicalType" />
+    /// </summary>
+    /// <seealso cref="Avro.Util.LogicalType" />
+    public class UnknownLogicalType : LogicalType
+    {
+        /// <summary>
+        /// Gets the schema.
+        /// </summary>
+        /// <value>The schema.</value>
+        public LogicalSchema Schema { get; }
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="UnknownLogicalType"/> class.
+        /// </summary>
+        /// <param name="schema">The schema.</param>
+        public UnknownLogicalType(LogicalSchema schema) : base(schema.LogicalTypeName)
+        {
+            this.Schema = schema;
+        }
+
+        /// <summary>
+        /// Converts a logical value to an instance of its base type.
+        /// </summary>
+        /// <param name="logicalValue">The logical value to convert.</param>
+        /// <param name="schema">The schema that represents the target of the conversion.</param>
+        /// <returns>An object representing the encoded value of the base type.</returns>
+        public override object ConvertToBaseValue(object logicalValue, LogicalSchema schema)
+        {
+            return logicalValue;
+        }
+
+        /// <summary>
+        /// Converts a base value to an instance of the logical type.
+        /// </summary>
+        /// <param name="baseValue">The base value to convert.</param>
+        /// <param name="schema">The schema that represents the target of the conversion.</param>
+        /// <returns>An object representing the encoded value of the logical type.</returns>
+        public override object ConvertToLogicalValue(object baseValue, LogicalSchema schema)
+        {
+            switch (schema.Name)
+            {
+                case @"string":
+                    return (System.String)baseValue;
+                case @"boolean":
+                    return (System.Boolean)baseValue;
+                case @"int":
+                    return (System.Int32)baseValue;
+                case @"long":
+                    return (System.Int64)baseValue;
+                case @"float":
+                    return (System.Single)baseValue;
+                case @"double":
+                    return (System.Double)baseValue;
+                case @"bytes":
+                    return (System.Byte[])baseValue;
+                default:
+                    return baseValue;
+            }
+        }
+
+        /// <summary>
+        /// Retrieve the .NET type that is represented by the logical type implementation.
+        /// </summary>
+        /// <param name="nullible">A flag indicating whether it should be nullible.</param>
+        /// <returns>Type.</returns>
+        public override Type GetCSharpType(bool nullible)
+        {
+            // handle all Primitive Types
+            switch (this.Schema.BaseSchema.Name)
+            {
+                case @"string":
+                    return typeof(System.String);
+                case @"boolean":
+                    return nullible ? typeof(System.Boolean?) : typeof(System.Boolean);
+                case @"int":
+                    return nullible ? typeof(System.Int32?) : typeof(System.Int32);
+                case @"long":
+                    return nullible ? typeof(System.Int64?) : typeof(System.Int64);
+                case @"float":
+                    return nullible ? typeof(System.Single?) : typeof(System.Single);
+                case @"double":
+                    return nullible ? typeof(System.Double?) : typeof(System.Double);
+                case @"bytes":
+                    return typeof(System.Byte[]);
+                default:
+                    return typeof(System.Object);
+            }
+        }
+
+        /// <summary>
+        /// Determines if a given object is an instance of the logical type.
+        /// </summary>
+        /// <param name="logicalValue">The logical value to test.</param>
+        /// <returns><c>true</c> if [is instance of logical type] [the specified logical value]; otherwise, <c>false</c>.</returns>
+        public override bool IsInstanceOfLogicalType(object logicalValue)
+        {
+            // handle all Primitive Types
+            switch (this.Schema.BaseSchema.Name)
+            {
+                case @"string":
+                    return logicalValue is System.String;
+                case @"boolean":
+                    return logicalValue is System.Boolean;
+                case @"int":
+                    return logicalValue is System.Int32;
+                case @"long":
+                    return logicalValue is System.Int64;
+                case @"float":
+                    return logicalValue is System.Single;
+                case @"double":
+                    return logicalValue is System.Double;
+                case @"bytes":
+                    return logicalValue is System.Byte[];
+                default:
+                    return true;
+            }
+        }
+
+    }
+}

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

@@ -606,9 +606,9 @@ public void GenerateSchemaWithNamespaceMapping(
             AvroGenHelper.TestSchema(schema, typeNamesToCheck, new Dictionary<string, string> { { namespaceMappingFrom, namespaceMappingTo } }, generatedFilesToCheck);
         }
 
-        [TestCase(_logicalTypesWithCustomConversion, typeof(AvroTypeException))]
-        [TestCase(_customConversionWithLogicalTypes, typeof(SchemaParseException))]
-        public void NotSupportedSchema(string schema, Type expectedException)
+        [TestCase(_logicalTypesWithCustomConversion)]
+        [TestCase(_customConversionWithLogicalTypes)]
+        public void UnknownLogicalTypesFallbackToBaseType(string schema)
         {
             // Create temp folder
             string outputDir = AvroGenHelper.CreateEmptyTemporaryFolder(out string uniqueId);
@@ -619,7 +619,9 @@ 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));
+                // Assert that the generator successfully runs (exit code 0)
+                // by ignoring the unknown logical types and using the underlying base types
+                Assert.That(AvroGenTool.GenSchema(schemaFileName, outputDir, new Dictionary<string, string>(), false), Is.EqualTo(0));
             }
             finally
             {

lang/csharp/src/apache/test/File/FileTests.cs

@@ -35,6 +35,34 @@ public class FileTests
         const string specificSchema  = "{\"type\":\"record\",\"name\":\"Foo\",\"namespace\":\"Avro.Test.File\",\"fields\":"
                                      + "[{\"name\":\"name\",\"type\":[\"null\",\"string\"]},{\"name\":\"age\",\"type\":\"int\"}]}";
 
+        /// <summary>
+        /// This test case added to confirm standalone serialization / deserialization behavior of new type UnknownLogicalType
+        /// </summary>
+        const string unknowLogicalTypeSchema = @"
+{
+	""type"" : ""record"",
+	""name"" : ""Foo"",
+	""namespace"" : ""Avro.Test.File"",
+	""fields"": [
+		{
+			""name"" :""name"",
+			""type"":  [
+                ""null"",
+                {
+                  ""logicalType"": ""varchar"",
+                  ""maxLength"": 65,
+                  ""type"": ""string""
+                }
+              ]
+		}, 
+		{
+			""name"" : ""age"",
+			""type"" : ""int""
+		}
+	]
+}
+";
+
         private static IEnumerable<TestCaseData> TestSpecificDataSource()
         {
             foreach (Codec.Type codecType in Enum.GetValues(typeof(Codec.Type)))
@@ -100,6 +128,11 @@ private static IEnumerable<TestCaseData> TestSpecificDataSource()
                         new object[] { "Bob", 9 },
                         new object[] { null, 48 }
                     }, codecType).SetName("{m}(Case3,{2})");
+
+                yield return new TestCaseData(unknowLogicalTypeSchema, new object[]
+                    {
+                        new object[] { "John", 23 }
+                    }, codecType).SetName("{m}(Case4,{2})");
             }
         }
 

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
 {
@@ -567,12 +568,80 @@ public void TestLogicalPrimitive(string s, string baseType, string logicalType)
             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 schema = Schema.Parse(s);
+            Assert.IsNotNull(schema);
+            Assert.IsInstanceOf(typeof(LogicalSchema), schema);
 
-            Assert.AreEqual("Logical type '" + unknownType + "' is not supported.", err.Message);
+            if (schema is LogicalSchema logicalSchema)
+            {
+                Assert.IsInstanceOf(typeof(UnknownLogicalType), logicalSchema.LogicalType);
+                Assert.AreEqual(logicalSchema.LogicalTypeName, unknownType);
+            }
+            else
+            {
+                Assert.Fail("Parsed schema was not a LogicalSchema");
+            }
+        }
+
+        /*
+            {
+              "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 = (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")]