Allow NullLogicalType given a PhysicalType (#302)

* Allow NullLogicalType given a PhysicalType

* Add a unit test for Null logical type roundtrip

* Only allow the null logical type for optional columns

* Parametrize the unit test for Null logical type roundtrip

Co-authored-by: Mark Sumner <msumner91@users.noreply.github.com>
Co-authored-by: Adam Reeve <adreeve@gmail.com>
Co-authored-by: Matthew Harward <mharward-gr@users.noreply.github.com>

csharp.test/TestCaseGenericAttribute.cs

@@ -0,0 +1,36 @@
+using System;
+using System.Collections.Generic;
+using NUnit.Framework;
+using NUnit.Framework.Interfaces;
+using NUnit.Framework.Internal;
+using NUnit.Framework.Internal.Builders;
+
+namespace ParquetSharp.Test
+{
+    [AttributeUsage(AttributeTargets.Method, AllowMultiple = true)]
+    internal sealed class TestCaseGenericAttribute : TestCaseAttribute, ITestBuilder
+    {
+        public TestCaseGenericAttribute(params object[] arguments)
+            : base(arguments)
+        {
+        }
+
+        public Type[] TypeArguments { get; set; } = null!;
+
+        IEnumerable<TestMethod> ITestBuilder.BuildFrom(IMethodInfo method, NUnit.Framework.Internal.Test? suite)
+        {
+            if (!method.IsGenericMethodDefinition)
+                return base.BuildFrom(method, suite);
+
+            if (TypeArguments == null || TypeArguments.Length != method.GetGenericArguments().Length)
+            {
+                var parms = new TestCaseParameters {RunState = RunState.NotRunnable};
+                parms.Properties.Set("_SKIPREASON", $"{nameof(TypeArguments)} should have {method.GetGenericArguments().Length} elements");
+                return new[] {new NUnitTestCaseBuilder().BuildTestMethod(method, suite, parms)};
+            }
+
+            var genMethod = method.MakeGenericMethod(TypeArguments);
+            return base.BuildFrom(genMethod, suite);
+        }
+    }
+}

csharp.test/TestLogicalTypeRoundtrip.cs

@@ -804,6 +804,43 @@ public static void TestRequiredStringRoundTrip()
             Assert.That(readValues, Is.EqualTo(stringValues));
         }
 
+        [TestCaseGeneric(PhysicalType.Int32, TypeArguments = new[] {typeof(int)})]
+        [TestCaseGeneric(PhysicalType.Int64, TypeArguments = new[] {typeof(long)})]
+        [TestCaseGeneric(PhysicalType.Int96, TypeArguments = new[] {typeof(Int96)})]
+        [TestCaseGeneric(PhysicalType.Boolean, TypeArguments = new[] {typeof(bool)})]
+        [TestCaseGeneric(PhysicalType.Float, TypeArguments = new[] {typeof(float)})]
+        [TestCaseGeneric(PhysicalType.Double, TypeArguments = new[] {typeof(double)})]
+        public static void TestNullLogicalTypeRoundTrip<T>(PhysicalType physicalType) where T : struct
+        {
+            var values = new T?[] {null, null};
+
+            using var nullType = LogicalType.Null();
+            using var nullColumn = new PrimitiveNode("nulls", Repetition.Optional, nullType, physicalType);
+            using var schemaNode = new GroupNode("schema", Repetition.Required, new[] {nullColumn});
+
+            using var buffer = new ResizableBuffer();
+            using (var output = new BufferOutputStream(buffer))
+            {
+                using var builder = new WriterPropertiesBuilder();
+                using var writerProperties = builder.Build();
+                using var fileWriter = new ParquetFileWriter(output, schemaNode, writerProperties);
+                using var rowGroupWriter = fileWriter.AppendBufferedRowGroup();
+
+                using var colWriter = rowGroupWriter.Column(0).LogicalWriter<T?>();
+                colWriter.WriteBatch(values);
+                fileWriter.Close();
+            }
+
+            using var input = new BufferReader(buffer);
+            using var fileReader = new ParquetFileReader(input);
+            using var rowGroupReader = fileReader.RowGroup(0);
+            using var colReader = rowGroupReader.Column(0).LogicalReader<T?>();
+
+            var readValues = colReader.ReadAll((int) rowGroupReader.MetaData.NumRows);
+
+            Assert.That(readValues, Is.EqualTo(values));
+        }
+
         private static ExpectedColumn[] CreateExpectedColumns()
         {
             return new[]

csharp/LogicalTypeFactory.cs

@@ -82,14 +82,26 @@ public virtual unsafe (Type physicalType, Type logicalType) GetSystemTypes(Colum
                 return (DefaultPhysicalTypeMapping[physicalType], match.Key);
             }
 
-            if (logicalType is NoneLogicalType)
+            if (logicalType is NoneLogicalType or NullLogicalType)
             {
+                if (!nullable && logicalType is NullLogicalType)
+                {
+                    throw new ArgumentOutOfRangeException(nameof(logicalType), "The null logical type may only be used with optional columns");
+                }
                 switch (physicalType)
                 {
                     case PhysicalType.Int32:
                         return (typeof(int), nullable ? typeof(int?) : typeof(int));
                     case PhysicalType.Int64:
                         return (typeof(long), nullable ? typeof(long?) : typeof(long));
+                    case PhysicalType.Int96:
+                        return (typeof(Int96), nullable ? typeof(Int96?) : typeof(Int96));
+                    case PhysicalType.Boolean:
+                        return (typeof(bool), nullable ? typeof(bool?) : typeof(bool));
+                    case PhysicalType.Float:
+                        return (typeof(float), nullable ? typeof(float?) : typeof(float));
+                    case PhysicalType.Double:
+                        return (typeof(double), nullable ? typeof(double?) : typeof(double));
                 }
             }