fix(streaming): convert progressive frame values to match column types

Progressive frames send decimals as strings but DataTable expects SqlDecimal, causing ArgumentException during row assignment.

Add FrameValueTypeConverter to handle type conversions before DataRow assignment in both streaming handlers.

Add InternalsVisibleTo for tests.

Author: Per Kops

src/Atc.Kusto/Atc.Kusto.csproj

@@ -29,6 +29,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <InternalsVisibleTo Include="Atc.Kusto.Tests" />
     <InternalsVisibleTo Include="DynamicProxyGenAssembly2" />
   </ItemGroup>
 

src/Atc.Kusto/Handlers/Internal/BufferedStreamingQueryHandler.cs

@@ -275,7 +275,15 @@ private void HandleTableFragment(
         var record = new object[tableFragmentFrame.FieldCount];
         while (tableFragmentFrame.GetNextRecord(record))
         {
-            dt.Rows.Add(record);
+            var newRow = dt.NewRow();
+
+            // Convert frame values to match column types (Kusto sends decimals as strings in progressive frames)
+            for (var i = 0; i < record.Length; i++)
+            {
+                newRow[i] = FrameValueTypeConverter.ConvertToColumnType(record[i], dt.Columns[i].DataType);
+            }
+
+            dt.Rows.Add(newRow);
         }
     }
 

src/Atc.Kusto/Handlers/Internal/StreamingQueryHandler.cs

@@ -164,7 +164,13 @@ private async IAsyncEnumerable<T> HandleTableFragment(
 
             // Create a new row from the known schema.
             var newRow = dt.NewRow();
-            newRow.ItemArray = (object[])record.Clone(); // ???
+
+            // Convert frame values to match column types (Kusto sends decimals as strings in progressive frames)
+            for (var i = 0; i < record.Length; i++)
+            {
+                newRow[i] = FrameValueTypeConverter.ConvertToColumnType(record[i], dt.Columns[i].DataType);
+            }
+
             dt.Rows.Add(newRow);
 
             if (tableKind == WellKnownDataSet.PrimaryResult)

src/Atc.Kusto/Utilities/Internal/FrameValueTypeConverter.cs

@@ -0,0 +1,62 @@
+namespace Atc.Kusto.Utilities.Internal;
+
+/// <summary>
+/// Provides utility methods for converting Kusto progressive frame values to match DataTable column types.
+/// </summary>
+/// <remarks>
+/// Kusto progressive frames may send values in wire format (e.g., decimals as strings)
+/// that need conversion to match the DataTable column types defined by the schema.
+/// </remarks>
+internal static class FrameValueTypeConverter
+{
+    /// <summary>
+    /// Converts a value from a Kusto progressive frame to match the target column type.
+    /// </summary>
+    /// <param name="value">The value from the progressive frame.</param>
+    /// <param name="targetType">The expected column type from the DataTable schema.</param>
+    /// <returns>
+    /// The value converted to the target type, or <see cref="DBNull.Value"/> if the value is null.
+    /// If conversion fails, returns the original value.
+    /// </returns>
+    /// <remarks>
+    /// This method handles several known type mismatches:
+    /// <list type="bullet">
+    ///   <item><description>String → SqlDecimal: Kusto sends decimal values as strings in progressive frames.</description></item>
+    ///   <item><description>Other mismatches: Uses <see cref="Convert.ChangeType(object, Type, IFormatProvider)"/> with invariant culture.</description></item>
+    /// </list>
+    /// </remarks>
+    public static object? ConvertToColumnType(
+        object? value,
+        Type targetType)
+    {
+        if (value is null or DBNull)
+        {
+            return DBNull.Value;
+        }
+
+        // If types already match, return as-is
+        if (value.GetType() == targetType)
+        {
+            return value;
+        }
+
+        // Handle string → SqlDecimal conversion (Kusto sends decimals as strings in progressive frames)
+        if (value is string stringValue &&
+            targetType == typeof(SqlDecimal) &&
+            decimal.TryParse(stringValue, NumberStyles.Any, CultureInfo.InvariantCulture, out var decimalValue))
+        {
+            return new SqlDecimal(decimalValue);
+        }
+
+        // For other type mismatches, try Convert.ChangeType
+        try
+        {
+            return Convert.ChangeType(value, targetType, CultureInfo.InvariantCulture);
+        }
+        catch
+        {
+            // If conversion fails, return original value and let DataTable handle it
+            return value;
+        }
+    }
+}

test/Atc.Kusto.Tests/Utilities/Internal/FrameValueTypeConverterTests.cs

@@ -0,0 +1,182 @@
+namespace Atc.Kusto.Tests.Utilities.Internal;
+
+public sealed class FrameValueTypeConverterTests
+{
+    [Fact]
+    public void ConvertToColumnType_Should_Return_DBNull_When_Value_Is_Null()
+    {
+        // Arrange
+        var targetType = typeof(string);
+
+        // Act
+        var result = FrameValueTypeConverter.ConvertToColumnType(null, targetType);
+
+        // Assert
+        Assert.Equal(DBNull.Value, result);
+    }
+
+    [Fact]
+    public void ConvertToColumnType_Should_Return_DBNull_When_Value_Is_DBNull()
+    {
+        // Arrange
+        object? value = DBNull.Value;
+        var targetType = typeof(int);
+
+        // Act
+        var result = FrameValueTypeConverter.ConvertToColumnType(value, targetType);
+
+        // Assert
+        Assert.Equal(DBNull.Value, result);
+    }
+
+    [Fact]
+    public void ConvertToColumnType_Should_Return_Same_Value_When_Types_Match()
+    {
+        // Arrange
+        const int value = 42;
+        var targetType = typeof(int);
+
+        // Act
+        var result = FrameValueTypeConverter.ConvertToColumnType(value, targetType);
+
+        // Assert
+        Assert.Equal(42, result);
+        Assert.IsType<int>(result);
+    }
+
+    [Theory]
+    [InlineData("123.45", 123.45)]
+    [InlineData("0.001", 0.001)]
+    [InlineData("999999.99", 999999.99)]
+    [InlineData("-456.78", -456.78)]
+    public void ConvertToColumnType_Should_Convert_String_To_SqlDecimal(
+        string stringValue,
+        decimal expectedDecimalValue)
+    {
+        // Arrange
+        var targetType = typeof(SqlDecimal);
+
+        // Act
+        var result = FrameValueTypeConverter.ConvertToColumnType(stringValue, targetType);
+
+        // Assert
+        Assert.NotNull(result);
+        Assert.IsType<SqlDecimal>(result);
+        var sqlDecimal = (SqlDecimal)result;
+        Assert.Equal(expectedDecimalValue, (decimal)sqlDecimal);
+    }
+
+    [Fact]
+    public void ConvertToColumnType_Should_Handle_Invalid_String_For_SqlDecimal()
+    {
+        // Arrange
+        const string value = "not-a-number";
+        var targetType = typeof(SqlDecimal);
+
+        // Act
+        var result = FrameValueTypeConverter.ConvertToColumnType(value, targetType);
+
+        // Assert - should return original value when conversion fails
+        Assert.Equal("not-a-number", result);
+    }
+
+    [Theory]
+    [InlineData("42", typeof(int), 42)]
+    [InlineData("3.14", typeof(double), 3.14)]
+    [InlineData("true", typeof(bool), true)]
+    public void ConvertToColumnType_Should_Use_ChangeType_For_Other_Conversions(
+        string stringValue,
+        Type targetType,
+        object expectedValue)
+    {
+        // Act
+        var result = FrameValueTypeConverter.ConvertToColumnType(stringValue, targetType);
+
+        // Assert
+        Assert.NotNull(result);
+        Assert.IsType(targetType, result);
+        Assert.Equal(expectedValue, result);
+    }
+
+    [Fact]
+    public void ConvertToColumnType_Should_Return_Original_Value_When_Conversion_Fails()
+    {
+        // Arrange - try to convert a complex object to an int
+        var value = new { Property = "test" };
+        var targetType = typeof(int);
+
+        // Act
+        var result = FrameValueTypeConverter.ConvertToColumnType(value, targetType);
+
+        // Assert - should return original value
+        Assert.Equal(value, result);
+    }
+
+    [Fact]
+    public void ConvertToColumnType_Should_Handle_Numeric_Type_Conversions()
+    {
+        // Arrange
+        const int value = 42;
+        var targetType = typeof(long);
+
+        // Act
+        var result = FrameValueTypeConverter.ConvertToColumnType(value, targetType);
+
+        // Assert
+        Assert.NotNull(result);
+        Assert.IsType<long>(result);
+        Assert.Equal(42L, result);
+    }
+
+    [Fact]
+    public void ConvertToColumnType_Should_Use_InvariantCulture()
+    {
+        // Arrange - use a decimal string with period (not comma) as decimal separator
+        const string value = "1234.56";
+        var targetType = typeof(decimal);
+
+        // Act
+        var result = FrameValueTypeConverter.ConvertToColumnType(value, targetType);
+
+        // Assert
+        Assert.NotNull(result);
+        Assert.IsType<decimal>(result);
+        Assert.Equal(1234.56m, result);
+    }
+
+    [Fact]
+    public void ConvertToColumnType_Should_Handle_String_To_DateTime()
+    {
+        // Arrange
+        const string value = "2024-01-15T10:30:00Z";
+        var targetType = typeof(DateTime);
+
+        // Act
+        var result = FrameValueTypeConverter.ConvertToColumnType(value, targetType);
+
+        // Assert
+        Assert.NotNull(result);
+        Assert.IsType<DateTime>(result);
+    }
+
+    [Theory]
+    [InlineData("20443.07")]
+    [InlineData("12345.6789")]
+    [InlineData("0.00")]
+    public void ConvertToColumnType_Should_Handle_Realistic_Kusto_Decimal_Strings(string decimalString)
+    {
+        // Arrange - this simulates the actual scenario from the Kusto progressive frames
+        var targetType = typeof(SqlDecimal);
+
+        // Act
+        var result = FrameValueTypeConverter.ConvertToColumnType(decimalString, targetType);
+
+        // Assert
+        Assert.NotNull(result);
+        Assert.IsType<SqlDecimal>(result);
+
+        var sqlDecimal = (SqlDecimal)result;
+        var expectedDecimal = decimal.Parse(decimalString, CultureInfo.InvariantCulture);
+        Assert.Equal(expectedDecimal, (decimal)sqlDecimal);
+    }
+}