DataAnalysisLibrary.Connectors.Adx 0.1.2

DataAnalysisLibrary

Current Version: 0.1.2 · 繁體中文

A clean, loosely-coupled C# data analysis library that abstracts query-language generation (KQL / T-SQL / PostgreSQL SQL) away from your application code. Query multiple heterogeneous data sources through a single unified interface — no database-specific logic leaks into your business layer.

Live Demo

Try it out at https://jackyai.azurewebsites.net/data-analysis

The demo runs all four query types live against a real SQL Server table (dbo.Orders). Sign in to add your own Azure Data Explorer or SQL Server connectors.

Installation

Install Core (required) plus whichever connector(s) you need:

# Required
dotnet add package DataAnalysisLibrary.Core

# One or more connectors
dotnet add package DataAnalysisLibrary.Connectors.Adx        # Azure Data Explorer (KQL)
dotnet add package DataAnalysisLibrary.Connectors.MsSql       # Microsoft SQL Server
dotnet add package DataAnalysisLibrary.Connectors.PostgreSql  # PostgreSQL

Quick Start

Step 1 — Build the engine

Register your data sources with DataAnalysisEngineBuilder and call Build(). The resulting engine is thread-safe and suitable for singleton injection.

// ADX (Service Principal auth)
var adxConfig = new AdxConfig
{
    SourceKey    = "adx",
    ClusterUri   = "https://yourcluster.kusto.windows.net",
    Database     = "YourDatabase",
    ClientId     = "your-client-id",
    ClientSecret = "your-client-secret",
    TenantId     = "your-tenant-id"
};

// SQL Server
var mssqlConfig = new MsSqlConfig
{
    SourceKey        = "mssql",
    ConnectionString = "Server=myserver;Database=mydb;Integrated Security=true;"
};

// PostgreSQL
var pgConfig = new PostgreSqlConfig
{
    SourceKey        = "pg",
    ConnectionString = "Host=localhost;Database=mydb;Username=postgres;Password=secret"
};

var engine = new DataAnalysisEngineBuilder()
    .AddSource(new AdxDataSourceFactory(adxConfig))
    .AddSource(new MsSqlDataSourceFactory(mssqlConfig))
    .AddSource(new PostgreSqlDataSourceFactory(pgConfig))
    .Build();

ASP.NET Core DI:

builder.Services.AddSingleton(engine);

Step 2 — Run a query

Pick one of the four query methods and pass the appropriate SourceKey.

Query Types

Summarize — Aggregation

Equivalent to SQL GROUP BY / ADX summarize. Use it to compute aggregated metrics.

var result = await engine.SummarizeAsync(new SummarizeRequest
{
    SourceKey = "mssql",
    TableName = "Orders",
    Filters   = [new FilterDimension("Region", ["APAC", "EMEA"])],
    ViewBy    = [new GroupField("Region"), new GroupField("ProductCategory")],
    Measures  =
    [
        new MeasureField("Revenue",    AggregationType.Sum),
        new MeasureField("OrderId",    AggregationType.Count,         Alias: "OrderCount"),
        new MeasureField("CustomerId", AggregationType.DistinctCount, Alias: "UniqueCustomers"),
        new MeasureField("UnitPrice",  AggregationType.Average,       Alias: "AvgPrice"),
    ]
});

if (result.IsSuccess)
{
    foreach (var row in result.Value!.Rows)
    {
        var region  = row.Get<string>("Region");
        var revenue = row.Get<decimal>("Revenue");
        var orders  = row.Get<int>("OrderCount");
        Console.WriteLine($"{region}: {revenue:C} ({orders} orders)");
    }
}

Available aggregation types:

ViewBy — Group Field Types

ViewBy accepts any mix of four GroupFieldBase subtypes. The library translates each one to the correct SQL / KQL expression per connector.

GroupField — plain column grouping

new GroupField("Region")               // GROUP BY [Region]
new GroupField("Region", Alias: "Rgn") // GROUP BY [Region] AS [Rgn]

DateTruncGroupField — date/time truncation

Groups a datetime column by a calendar granularity. No application-side processing needed — the truncation runs inside the query engine.

ViewBy =
[
    new DateTruncGroupField("OrderDate", DateGranularity.Month),         // "2024-03"
    new DateTruncGroupField("OrderDate", DateGranularity.Quarter, "Qtr") // "2024Q1"
]

Available granularities and generated expressions:

NumericBinGroupField — numeric range buckets

Splits a numeric column into labeled, left-closed right-open intervals [Min, Max). null on either end means open-ended.

ViewBy =
[
    new NumericBinGroupField("Age",
        Bins:
        [
            new NumericBin(null, 18,   "Minor"),   // Age < 18
            new NumericBin(18,   65,   "Adult"),   // 18 ≤ Age < 65
            new NumericBin(65,   null, "Senior"),  // Age ≥ 65
        ],
        Alias:    "AgeGroup",
        Fallback: "Unknown")   // rows matching no bin → "Unknown" (default: "Other")
]

Generated SQL (T-SQL / PostgreSQL):

CASE
  WHEN [Age] < 18        THEN 'Minor'
  WHEN [Age] >= 18 AND [Age] < 65 THEN 'Adult'
  WHEN [Age] >= 65       THEN 'Senior'
  ELSE 'Unknown'
END AS [AgeGroup]

Generated KQL:

['AgeGroup']=case(['Age'] < 18, 'Minor', ['Age'] >= 18 and ['Age'] < 65, 'Adult', ['Age'] >= 65, 'Senior', 'Unknown')

StringGroupingField — many-to-one string mapping

Consolidates raw string values into logical groups (e.g. roll up countries → regions). Values not matched by any group receive the Fallback label.

ViewBy =
[
    new StringGroupingField("Country",
        Groups:
        [
            new StringGroup("AMEA",    ["TW", "HK", "SG", "IN"]),
            new StringGroup("America", ["US", "CA", "BR"]),
        ],
        Alias:    "Region",
        Fallback: "Other")
]

Generated SQL (T-SQL / PostgreSQL):

CASE
  WHEN [Country] IN ('TW', 'HK', 'SG', 'IN') THEN 'AMEA'
  WHEN [Country] IN ('US', 'CA', 'BR')        THEN 'America'
  ELSE 'Other'
END AS [Region]

Generated KQL:

['Region']=case(['Country'] in ('TW', 'HK', 'SG', 'IN'), 'AMEA', ['Country'] in ('US', 'CA', 'BR'), 'America', 'Other')

Mixing GroupField types

All four types can be combined freely in one ViewBy list:

var result = await engine.SummarizeAsync(new SummarizeRequest
{
    SourceKey = "mssql",
    TableName = "Orders",
    ViewBy =
    [
        new StringGroupingField("Country", [
            new StringGroup("AMEA",    ["TW", "HK"]),
            new StringGroup("America", ["US", "CA"]),
        ], Alias: "Region"),
        new DateTruncGroupField("OrderDate", DateGranularity.Month),
        new NumericBinGroupField("Amount", [
            new NumericBin(null, 1000,  "Small"),
            new NumericBin(1000, 10000, "Medium"),
            new NumericBin(10000, null, "Large"),
        ], Alias: "DealSize"),
    ],
    Measures = [new MeasureField("OrderId", AggregationType.Count, Alias: "Orders")]
});

Details — Row-level data

Retrieves raw rows with server-side pagination — no in-memory buffering.

var result = await engine.DetailsAsync(new DetailsRequest
{
    SourceKey    = "adx",
    TableName    = "EventLogs",
    SelectFields = ["EventId", "Timestamp", "Level", "Message"],  // empty = all columns
    Filters      =
    [
        new FilterDimension("Level",  ["ERROR", "CRITICAL"]),
        new FilterDimension("Region", ["TW"], FilterOperator.NotIn),
    ],
    OrderBy = [new SortField("Timestamp", Descending: true)],
    Skip    = 0,
    Take    = 50,
});

if (result.IsSuccess)
{
    Console.WriteLine($"Total: {result.Value!.TotalCount} rows");
    foreach (var row in result.Value.Rows)
        Console.WriteLine($"[{row["Level"]}] {row["Message"]}");
}

FieldList — Schema inspection

Returns column definitions including the native type and a cross-source canonical type — useful for dynamic UI rendering.

var result = await engine.FieldListAsync(new FieldListRequest
{
    SourceKey = "pg",
    TableName = "customers"
});

if (result.IsSuccess)
{
    foreach (var field in result.Value!.Fields)
    {
        // DataType      = native type string, e.g. "character varying", "integer"
        // CanonicalType = unified enum across all connectors, e.g. String, Integer
        Console.WriteLine($"{field.Name,-20} {field.DataType,-25} → {field.CanonicalType}  nullable={field.IsNullable}");
    }
}

Example output (PostgreSQL customers table):

id                   integer                   → Integer        nullable=False
name                 character varying         → String         nullable=False
email                character varying         → String         nullable=True
created_at           timestamp with time zone  → DateTimeOffset nullable=False
is_active            boolean                   → Boolean        nullable=False

Mapping CanonicalType to UI components

foreach (var field in result.Value!.Fields)
{
    var component = field.CanonicalType switch
    {
        ColumnDataType.DateTime or
        ColumnDataType.DateTimeOffset or
        ColumnDataType.Date    => "DatePicker",

        ColumnDataType.Boolean => "Toggle",

        ColumnDataType.Integer or
        ColumnDataType.Float   or
        ColumnDataType.Decimal => "NumberInput",

        ColumnDataType.Json    => "JsonEditor",

        _                      => "TextInput"   // String, Unknown, etc.
    };

    Console.WriteLine($"{field.Name} → {component}");
}

Full ColumnDataType reference:

MaxDate — Date boundary

Returns MAX() of a date column. Commonly used to check data freshness or compute time ranges.

var result = await engine.MaxDateAsync(new MaxDateRequest
{
    SourceKey     = "adx",
    TableName     = "TelemetryEvents",
    DateFieldName = "Timestamp",
    Filters       = [new FilterDimension("DeviceType", ["Sensor"])]
});

if (result.IsSuccess)
{
    var bound = result.Value!.BoundDate;
    if (bound is null)
        Console.WriteLine("Table is empty");
    else
        Console.WriteLine($"Latest record: {bound:yyyy-MM-dd HH:mm:ss zzz}");
}

Filters (FilterDimension)

// IN (default) — most common
new FilterDimension("Status", ["Active", "Pending"])

// NOT IN
new FilterDimension("Region", ["TW"], FilterOperator.NotIn)

// Single-value comparisons
new FilterDimension("Score",    ["90"], FilterOperator.GreaterThan)
new FilterDimension("Priority", ["3"],  FilterOperator.LessThan)
new FilterDimension("Code",     ["US"], FilterOperator.Equals)

// Range (Between requires exactly two values: [min, max])
new FilterDimension("Amount", ["100", "500"], FilterOperator.Between)

// Substring match (translated to LIKE / ILIKE)
new FilterDimension("Name", ["John"], FilterOperator.Contains)

ADX limitation: ADX only supports In and NotIn. Other operators will throw NotSupportedException.

Result Pattern

All query methods return QueryResult<T> and never throw (except OperationCanceledException).

var result = await engine.SummarizeAsync(request);

if (result.IsSuccess)
{
    var data = result.Value!;  // guaranteed non-null
    // use data...
}
else
{
    logger.LogError("Query failed [{Code}]: {Message}", result.ErrorCode, result.ErrorMessage);
}

Built-in error codes:

CancellationToken Support

All query methods accept a CancellationToken, making them suitable for ASP.NET Core request-abort scenarios:

public async Task<IActionResult> GetSummary(CancellationToken ct)
{
    var result = await engine.SummarizeAsync(request, ct);
    // If the client disconnects, OperationCanceledException is thrown;
    // ASP.NET Core handles it automatically.
    ...
}

Extending with a New Connector

Add new data sources without modifying any existing code (Open/Closed Principle):

var engine = new DataAnalysisEngineBuilder()
    .AddSource(new AdxDataSourceFactory(adxConfig))
    .AddSource(new MyCustomDataSourceFactory(myConfig))  // ← just add this
    .Build();

See the Developer Guide for implementation details.

Requirements

• .NET 8.0+
• C# 12+