[ksqlDb.RestApi.Client]: added AsStruct to Fluent API

Author: tomasfabian

ksqlDb.RestApi.Client/FluentAPI/Builders/FieldTypeBuilder.cs

@@ -26,6 +26,12 @@ public interface IFieldTypeBuilder<TProperty>
     /// <param name="columnName">The name of the column in the record schema.</param>
     /// <returns>The same <see cref="IFieldTypeBuilder{TProperty}"/> instance so that multiple calls can be chained.</returns>
     IFieldTypeBuilder<TProperty> HasColumnName(string columnName);
+
+    /// <summary>
+    /// Marks the field as a ksqldb STRUCT type.
+    /// </summary>
+    /// <returns>The field type builder for chaining additional configuration.</returns>
+    IFieldTypeBuilder<TProperty> AsStruct();
   }
 
   internal class FieldTypeBuilder<TProperty>(FieldMetadata fieldMetadata)
@@ -37,6 +43,12 @@ public IFieldTypeBuilder<TProperty> HasColumnName(string columnName)
       return this;
     }
 
+    public IFieldTypeBuilder<TProperty> AsStruct()
+    {
+      fieldMetadata.IsStruct = true;
+      return this;
+    }
+
     public IFieldTypeBuilder<TProperty> Ignore()
     {
       fieldMetadata.Ignore = true;

ksqlDb.RestApi.Client/KSql/RestApi/Generators/TypeGenerator.cs

@@ -11,8 +11,6 @@ namespace ksqlDB.RestApi.Client.KSql.RestApi.Generators;
 
 internal sealed class TypeGenerator(IMetadataProvider metadataProvider) : EntityInfo(metadataProvider)
 {
-  private readonly KSqlTypeTranslator typeTranslator = new(metadataProvider);
-
   internal string Print<T>(TypeProperties properties)
   {
     StringBuilder stringBuilder = new();
@@ -30,11 +28,13 @@ private void PrintProperties<T>(StringBuilder stringBuilder, IdentifierEscaping
   {
     var ksqlProperties = new List<string>();
 
+    KSqlTypeTranslator<T> typeTranslator = new(metadataProvider);
+
     foreach (var memberInfo in Members<T>())
     {
       var type = GetMemberType(memberInfo);
 
-      var ksqlType = typeTranslator.Translate(type, escaping);
+      var ksqlType = typeTranslator.Translate(type, memberInfo, escaping);
 
       var memberName = memberInfo.GetMemberName(metadataProvider);
       var columnDefinition = $"{EscapeName(memberName, escaping)} {ksqlType}{typeTranslator.ExploreAttributes(typeof(T), memberInfo, type)}";

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

@@ -13,8 +13,6 @@ internal sealed class CreateEntity(IMetadataProvider metadataProvider) : EntityI
 {
   private readonly StringBuilder stringBuilder = new();
 
-  private readonly KSqlTypeTranslator typeTranslator = new(metadataProvider);
-
   internal string Print<T>(StatementContext statementContext, EntityCreationMetadata metadata, bool? ifNotExists)
   {
     stringBuilder.Clear();
@@ -42,12 +40,13 @@ internal string Print<T>(StatementContext statementContext, EntityCreationMetada
   private void PrintProperties<T>(StatementContext statementContext, EntityCreationMetadata metadata)
   {
     var ksqlProperties = new List<string>();
+    KSqlTypeTranslator<T> typeTranslator = new(metadataProvider);
 
     foreach (var memberInfo in Members<T>(metadata.IncludeReadOnlyProperties))
     {
       var type = GetMemberType(memberInfo);
 
-      var ksqlType = typeTranslator.Translate(type, metadata.IdentifierEscaping);
+      var ksqlType = typeTranslator.Translate(type, memberInfo, metadata.IdentifierEscaping);
 
       var columnName = IdentifierUtil.Format(memberInfo, metadata.IdentifierEscaping, metadataProvider);
       string columnDefinition = $"\t{columnName} {ksqlType}{typeTranslator.ExploreAttributes(typeof(T), memberInfo, type)}";

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

@@ -9,12 +9,12 @@
 
 namespace ksqlDB.RestApi.Client.KSql.RestApi.Statements
 {
-  internal sealed class KSqlTypeTranslator(IMetadataProvider metadataProvider) : EntityInfo(metadataProvider)
+  internal sealed class KSqlTypeTranslator<TEntity>(IMetadataProvider metadataProvider) : EntityInfo(metadataProvider)
   {
     private readonly IMetadataProvider metadataProvider = metadataProvider;
     private readonly DecimalTypeTranslator decimalTypeTranslator = new(metadataProvider);
 
-    internal string Translate(Type type, IdentifierEscaping escaping = IdentifierEscaping.Never)
+    internal string Translate(Type type, MemberInfo? memberInfo = null, IdentifierEscaping escaping = IdentifierEscaping.Never)
     {
       var ksqlType = string.Empty;
 
@@ -25,16 +25,16 @@ internal string Translate(Type type, IdentifierEscaping escaping = IdentifierEsc
         var elementType = type.GetElementType();
         if (elementType == null)
           throw new InvalidOperationException(nameof(elementType));
-        var elementTypeName = Translate(elementType, escaping);
+        var elementTypeName = Translate(elementType, memberInfo, escaping);
 
         ksqlType = $"{KSqlTypes.Array}<{elementTypeName}>";
       }
       else if (type.IsDictionary())
       {
         Type[] typeParameters = type.GetGenericArguments();
 
-        var keyType = Translate(typeParameters[0], escaping);
-        var valueType = Translate(typeParameters[1], escaping);
+        var keyType = Translate(typeParameters[0], memberInfo, escaping);
+        var valueType = Translate(typeParameters[1], memberInfo, escaping);
 
         ksqlType = $"{KSqlTypes.Map}<{keyType}, {valueType}>";
       }
@@ -58,7 +58,7 @@ internal string Translate(Type type, IdentifierEscaping escaping = IdentifierEsc
         ksqlType = KSqlTypes.Time;
       else if (type == typeof(DateTimeOffset))
         ksqlType = KSqlTypes.Timestamp;
-      else if (!type.IsGenericType && type.TryGetAttribute<StructAttribute>() != null)
+      else if (!type.IsGenericType && IsStructType(type, memberInfo))
       {
         var ksqlProperties = GetProperties(type, escaping);
 
@@ -88,7 +88,7 @@ internal string Translate(Type type, IdentifierEscaping escaping = IdentifierEsc
 
         if (elementType != null)
         {
-          string ksqlElementType = Translate(elementType, escaping);
+          string ksqlElementType = Translate(elementType, memberInfo, escaping);
 
           ksqlType = $"{KSqlTypes.Array}<{ksqlElementType}>";
         }
@@ -97,6 +97,22 @@ internal string Translate(Type type, IdentifierEscaping escaping = IdentifierEsc
       return ksqlType;
     }
 
+    private bool IsStructType(Type type, MemberInfo? memberInfo)
+    {
+      if (type.TryGetAttribute<StructAttribute>() != null)
+        return true;
+
+      if (memberInfo == null)
+        return false;
+
+      var entityMetadata = metadataProvider.GetEntities().FirstOrDefault(c => c.Type == typeof(TEntity));
+      var fieldMetadata = entityMetadata?.GetFieldMetadataBy(memberInfo);
+      return fieldMetadata is
+      {
+        IsStruct: true
+      };
+    }
+
     internal IEnumerable<string> GetProperties(Type type, IdentifierEscaping escaping)
     {
       var ksqlProperties = new List<string>();
@@ -105,7 +121,7 @@ internal IEnumerable<string> GetProperties(Type type, IdentifierEscaping escapin
       {
         var memberType = GetMemberType(memberInfo);
 
-        var ksqlType = Translate(memberType, escaping);
+        var ksqlType = Translate(memberType, memberInfo, escaping);
 
         string columnDefinition = $"{memberInfo.Format(escaping, metadataProvider as ModelBuilder)} {ksqlType}{ExploreAttributes(type, memberInfo, memberType)}";
 

ksqlDb.RestApi.Client/Metadata/FieldMetadata.cs

@@ -7,6 +7,7 @@ internal record FieldMetadata
     internal MemberInfo MemberInfo { get; init; } = null!;
     public bool Ignore { get; internal set; }
     public bool HasHeaders { get; internal set; }
+    internal bool IsStruct { get; set; }
     internal string Path { get; init; } = null!;
     internal string FullPath { get; init; } = null!;
     public string? ColumnName { get; set; }