Merge pull request #115 from vidyaranya92/feat/nullable-inserts

fix: casting nulls while generating insert statement

Author: tomasfabian

Tests/ksqlDB.RestApi.Client.Tests/KSql/RestApi/KSqlDbRestApiClientTests.cs

@@ -589,12 +589,14 @@ public class POC
     public int Id { get; init; }
     public string? Description { get; init; }
     public POC2[]? Entities { get; init; }
+    public int? Foo { get; init; }
   }
 
   public class POC2
   {
     public POC3[]? Poc3 { get; set; }
     public POC4[]? Poc4 { get; set; }
+    public POC5? Poc5 { get; set; }
   }
 
   public class POC3
@@ -606,13 +608,24 @@ public class POC4
   {
     public string? Description { get; init; }
   }
+  public class POC5
+  {
+    public double? Amount { get; init; }
+  }
 
   public static IEnumerable<(POC, string)> NullTestCases()
-  { // empty array constructors are invalid
-    yield return (new POC { Id = 1 }, "INSERT INTO POCS (Id, Description, Entities) VALUES (1, NULL, NULL);");
-    yield return (new POC { Id = 1, Entities = new POC2[0] }, "INSERT INTO POCS (Id, Description, Entities) VALUES (1, NULL, ARRAY_REMOVE(ARRAY[0], 0));");
-    yield return (new POC { Id = 1, Entities = new POC2[] { new POC2 { Poc3 = new POC3[0] } } }, "INSERT INTO POCS (Id, Description, Entities) VALUES (1, NULL, ARRAY[STRUCT(Poc3 := ARRAY_REMOVE(ARRAY[0], 0), Poc4 := ARRAY_REMOVE(ARRAY[0], 0))]);");
-    yield return (new POC { Id = 1, Entities = new POC2[] { new POC2 { Poc3 = new POC3[0], Poc4 = new POC4[0] } } }, "INSERT INTO POCS (Id, Description, Entities) VALUES (1, NULL, ARRAY[STRUCT(Poc3 := ARRAY_REMOVE(ARRAY[0], 0), Poc4 := ARRAY_REMOVE(ARRAY[0], 0))]);");
+  {
+    // empty array constructors are invalid
+    yield return (new POC { Id = 1, Foo = null}, "INSERT INTO POCS (Id, Description, Entities, Foo) VALUES (1, NULL, NULL, NULL);");
+    yield return (new POC { Id = 1, Entities = new POC2[0] }, "INSERT INTO POCS (Id, Description, Entities, Foo) VALUES (1, NULL, ARRAY_REMOVE(ARRAY[0], 0), NULL);");
+    yield return (new POC { Id = 1, Entities = new POC2[] { new POC2 { Poc3 = new POC3[0]} } },
+      "INSERT INTO POCS (Id, Description, Entities, Foo) VALUES (1, NULL, ARRAY[STRUCT(Poc3 := ARRAY_REMOVE(ARRAY[0], 0), Poc4 := ARRAY_REMOVE(ARRAY[0], 0), Poc5 := CAST(NULL AS STRUCT<Amount DOUBLE>))], NULL);");
+    yield return (new POC { Id = 1, Entities = new POC2[] { new POC2 { Poc3 = new POC3[0], Poc4 = new POC4[0] } } },
+      "INSERT INTO POCS (Id, Description, Entities, Foo) VALUES (1, NULL, ARRAY[STRUCT(Poc3 := ARRAY_REMOVE(ARRAY[0], 0), Poc4 := ARRAY_REMOVE(ARRAY[0], 0), Poc5 := CAST(NULL AS STRUCT<Amount DOUBLE>))], NULL);");
+    yield return (new POC { Id = 1, Entities = new POC2[] { new POC2 { Poc3 = new[] { new POC3() {Description = null}}, Poc4 = new POC4[0] } } },
+      "INSERT INTO POCS (Id, Description, Entities, Foo) VALUES (1, NULL, ARRAY[STRUCT(Poc3 := ARRAY[STRUCT(Description := CAST(NULL AS VARCHAR))], Poc4 := ARRAY_REMOVE(ARRAY[0], 0), Poc5 := CAST(NULL AS STRUCT<Amount DOUBLE>))], NULL);");
+    yield return (new POC { Id = 1, Entities = new POC2[] { new POC2 { Poc3 = new[] { new POC3() {Description = null}}, Poc4 = new POC4[0], Poc5 = new POC5() {Amount = 10.05}} } },
+      "INSERT INTO POCS (Id, Description, Entities, Foo) VALUES (1, NULL, ARRAY[STRUCT(Poc3 := ARRAY[STRUCT(Description := CAST(NULL AS VARCHAR))], Poc4 := ARRAY_REMOVE(ARRAY[0], 0), Poc5 := STRUCT(Amount := 10.05))], NULL);");
   }
 
   [TestCaseSource(nameof(NullTestCases))]
@@ -623,6 +636,7 @@ public void ToInsertStatement_WithNullHandling((POC, string) testCase)
     var modelBuilder = new ModelBuilder();
     modelBuilder.Entity<POC>().HasKey(i => i.Id);
     modelBuilder.Entity<POC>().Property(c => c.Entities).AsStruct();
+    modelBuilder.Entity<POC>().Property(c => c.Entities!.FirstOrDefault()!.Poc5).AsStruct();
     var sut = new KSqlDbRestApiClient(HttpClientFactory, modelBuilder, LoggerFactoryMock.Object);
 
     //Act

ksqlDb.RestApi.Client/ChangeLog.md

@@ -1,5 +1,10 @@
 # ksqlDB.RestApi.Client
 
+# 7.1.5
+
+## 🐛 Bug Fixes
+- casting nulls while generating insert statement. #115 (contributed by @vidyaranya92)
+
 # 7.1.4
 
 ## 🐛 Bug Fixes
@@ -80,7 +85,7 @@ This enables updating the credentials at runtime by setting them on the `KSqlDbC
 ## 🚀 New Features
 1. **KSqlDbRestApiClient Constructor Update**:
     - The `KSqlDbRestApiClient` class constructor now includes a parameter for `KSqlDBRestApiClientOptions`.
-    
+
 2. **EntityCreationMetadata Update**:
     - The `EntityCreationMetadata.ShouldPluralizeEntityName` property was modified to be a nullable boolean (`bool?`), and the default value of `true` was removed.
 
@@ -91,7 +96,7 @@ This enables updating the credentials at runtime by setting them on the `KSqlDbC
         - `InsertProperties`
         - `DropFromItemProperties`
     - If `ShouldPluralizeEntityName` is null, the methods will set it using the value from the `KSqlDBRestApiClientOptions`.
-    
+
 ## 🐛 Bug Fix
 - `KSqlDBContext` removed `!NETSTANDARD` pragma from `OnDisposeAsync`
 
@@ -124,7 +129,7 @@ This enables updating the credentials at runtime by setting them on the `KSqlDbC
 - see also [breakingchanges.md](https://github.com/tomasfabian/ksqlDB.RestApi.Client-DotNet/blob/main/docs/breaking_changes.md#v600)
 
 ## 🐛 Bug Fix
-- CreateQueryStream doesn't always use configured parameters. The PullQuery functionality was overriding the options configured for the PushQuery. #75 reported by @jbkuczma 
+- CreateQueryStream doesn't always use configured parameters. The PullQuery functionality was overriding the options configured for the PushQuery. #75 reported by @jbkuczma
 
 # 5.1.0
 - added `InsertIntoAsync` Qbservable extension for executing `INSERT INTO <stream-name> SELECT` statements.

ksqlDb.RestApi.Client/KSql/RestApi/Statements/CreateKSqlValue.cs

@@ -15,7 +15,7 @@ internal sealed class CreateKSqlValue(IMetadataProvider metadataProvider) : Enti
 {
   private const string CREATE_EMPTY_ARRAY = "ARRAY_REMOVE(ARRAY[0], 0)";
 
-  public object ExtractValue<T>(T inputValue, IValueFormatters valueFormatters, MemberInfo memberInfo, Type type, Func<MemberInfo, string> formatter)
+  public object ExtractValue<T>(T inputValue, IValueFormatters valueFormatters, MemberInfo memberInfo, Type type, Func<MemberInfo, string> formatter, bool castNulls = false)
   {
     Type valueType = inputValue.GetType();
 
@@ -27,7 +27,15 @@ public object ExtractValue<T>(T inputValue, IValueFormatters valueFormatters, Me
       value = valueType.GetField(memberInfo.Name)?.GetValue(inputValue);
 
     if (value == null)
+    {
+      if (castNulls && !type.IsArray)
+      {
+        KSqlTypeTranslator<T> typeTranslator = new(metadataProvider);
+        var ksqlType = typeTranslator.Translate(type, memberInfo);
+        return $"CAST({KSqlTypes.Null} AS {ksqlType})";
+      }
       return KSqlTypes.Null;
+    }
 
     if (type == typeof(decimal))
     {
@@ -148,7 +156,7 @@ private void GenerateStruct<T>(IValueFormatters valueFormatters, Type type, Func
 
       type = GetMemberType(memberInfo2);
 
-      var innerValue = ExtractValue(value, valueFormatters, memberInfo2, type, formatter);
+      var innerValue = ExtractValue(value, valueFormatters, memberInfo2, type, formatter, true);
       var name = formatter(memberInfo2);
       if (type.IsArray && KSqlTypes.Null.Equals(innerValue))
         sb.Append($"{name} := {CREATE_EMPTY_ARRAY}");

ksqlDb.RestApi.Client/KSql/RestApi/Statements/EntityInfo.cs

@@ -66,12 +66,17 @@ protected bool IsStructType<TEntity>(Type type, MemberInfo? memberInfo)
     if (memberInfo == null)
       return false;
 
+    // First check if the exact type is directly marked as struct in metadata
     var entityMetadata = metadataProvider.GetEntities().FirstOrDefault(c => c.Type == typeof(TEntity));
     var fieldMetadata = entityMetadata?.GetFieldMetadataBy(memberInfo);
-    return fieldMetadata is
-    {
-      IsStruct: true
-    };
+
+    if (fieldMetadata is { IsStruct: true })
+      return true;
+
+    // Check all entities for any struct field using this type
+    return metadataProvider.GetEntities()
+      .SelectMany(metadata => metadata.FieldsMetadataDict.Values)
+      .Any(field => field.IsStruct && GetMemberType(field.MemberInfo) == type);
   }
 
   protected static Type GetMemberType(MemberInfo memberInfo)

ksqlDb.RestApi.Client/ksqlDb.RestApi.Client.csproj

@@ -15,8 +15,8 @@
             Documentation for the library can be found at https://github.com/tomasfabian/ksqlDB.RestApi.Client-DotNet/blob/main/README.md.
         </Description>
         <PackageTags>ksql ksqlDB LINQ .NET csharp push query</PackageTags>
-        <Version>7.1.4</Version>
-        <AssemblyVersion>7.1.4.0</AssemblyVersion>
+        <Version>7.1.5</Version>
+        <AssemblyVersion>7.1.5.0</AssemblyVersion>
         <LangVersion>13.0</LangVersion>
         <ImplicitUsings>enable</ImplicitUsings>
         <PackageReleaseNotes>https://github.com/tomasfabian/ksqlDB.RestApi.Client-DotNet/blob/main/ksqlDb.RestApi.Client/ChangeLog.md</PackageReleaseNotes>