ClickHouse C# client

dotnet add package ClickHouse.Driver

Install-Package ClickHouse.Driver

using ClickHouse.Driver;

// クライアントを作成する（通常はシングルトンとして）
using var client = new ClickHouseClient("Host=my.clickhouse;Protocol=https;Port=8443;Username=user");

// クエリを実行する
var version = await client.ExecuteScalarAsync("SELECT version()");
Console.WriteLine(version);

Host=localhost;Port=8123;Username=default;Password=secret;Database=mydb

Host=localhost;set_max_threads=4;set_readonly=1;set_max_memory_usage=10000000000

var options = new QueryOptions
{
    QueryId = "report-2024-001",
    Database = "analytics",
    CustomSettings = new Dictionary<string, object>
    {
        { "max_threads", 4 },
        { "max_memory_usage", 10_000_000_000 }
    },
    MaxExecutionTime = TimeSpan.FromMinutes(5)
};

var reader = await client.ExecuteReaderAsync(
    "SELECT * FROM large_table",
    parameters: null,
    options: options
);

var insertOptions = new InsertOptions
{
    BatchSize = 50_000,
    MaxDegreeOfParallelism = 4,
    QueryId = "bulk-import-001"
};

long rowsInserted = await client.InsertBinaryAsync(
    "my_table",
    columns,
    rows,
    insertOptions
);

var options = new InsertOptions
{
    ColumnTypes = new Dictionary<string, string>
    {
        ["id"] = "UInt64",
        ["name"] = "Nullable(String)",
        ["score"] = "Float32",
    },
};

await client.InsertBinaryAsync("my_table", ["id", "name", "score"], rows, options);

var options = new InsertOptions { UseSchemaCache = true };

// 最初の呼び出しでサーバーからスキーマを取得する
await client.InsertBinaryAsync("my_table", columns, batch1, options);

// 2回目の呼び出しではキャッシュ済みスキーマを再利用する — 追加のラウンドトリップなし
await client.InsertBinaryAsync("my_table", columns, batch2, options);

using ClickHouse.Driver;

using var client = new ClickHouseClient("Host=localhost;Username=default;Password=secret");

using ClickHouse.Driver;

var settings = new ClickHouseClientSettings
{
    Host = "localhost",
    Username = "default",
    Password = "secret"
};
using var client = new ClickHouseClient(settings);

// In your DI configuration. No AutomaticDecompression needed — the driver decodes
// compressed responses itself, and a mask here would widen its Accept-Encoding.
services.AddHttpClient("ClickHouse", client =>
{
    client.Timeout = TimeSpan.FromMinutes(5);
});

// Create client with factory
var factory = serviceProvider.GetRequiredService<IHttpClientFactory>();
var client = new ClickHouseClient("Host=localhost", factory, "ClickHouse");

// テーブルを作成する
await client.ExecuteNonQueryAsync(
    "CREATE TABLE IF NOT EXISTS default.my_table (id Int64, name String) ENGINE = Memory"
);

// テーブルを削除する
await client.ExecuteNonQueryAsync("DROP TABLE IF EXISTS default.my_table");

var count = await client.ExecuteScalarAsync("SELECT count() FROM default.my_table");
Console.WriteLine($"行数: {count}");

var version = await client.ExecuteScalarAsync("SELECT version()");
Console.WriteLine($"サーバーバージョン: {version}");

using ClickHouse.Driver;
using ClickHouse.Driver.ADO.Parameters;

var parameters = new ClickHouseParameterCollection();
parameters.AddParameter("id", 1L);
parameters.AddParameter("name", "Alice");

await client.ExecuteNonQueryAsync(
    "INSERT INTO default.my_table (id, name) VALUES ({id:Int64}, {name:String})",
    parameters
);

// IEnumerable<object[]> としてデータを準備する
var rows = Enumerable.Range(0, 1_000_000)
    .Select(i => new object[] { (long)i, $"value{i}" });

var columns = new[] { "id", "name" };

// 基本的な挿入
long rowsInserted = await client.InsertBinaryAsync("default.my_table", columns, rows);
Console.WriteLine($"Rows inserted: {rowsInserted}");

var options = new InsertOptions
{
    BatchSize = 100_000,           // バッチあたりの行数（デフォルト: 100,000）
    MaxDegreeOfParallelism = 4     // 並列バッチアップロード数（デフォルト: 1）
};

// テーブルのカラムに対応するPOCOを定義する
public class SensorReading
{
    public ulong Id { get; set; }
    public string SensorName { get; set; }
    public double Value { get; set; }
    public DateTime Timestamp { get; set; }
}

// 型を登録する（クライアントのライフタイムにつき1回）
client.RegisterBinaryInsertType<SensorReading>();

// 直接挿入する — カラム名はプロパティ名から導出される
var readings = Enumerable.Range(0, 100_000)
    .Select(i => new SensorReading
    {
        Id = (ulong)i,
        SensorName = $"sensor_{i % 10}",
        Value = Random.Shared.NextDouble() * 100,
        Timestamp = DateTime.UtcNow,
    });

long rowsInserted = await client.InsertBinaryAsync("sensors", readings);

public class Event
{
    [ClickHouseColumn(Name = "event_id")]     // 異なる名前のカラムにマッピングする
    public ulong Id { get; set; }

    [ClickHouseColumn(Type = "LowCardinality(String)")]  // ClickHouseの型を明示的に指定
    public string Category { get; set; }

    public string Payload { get; set; }

    [ClickHouseNotMapped]                     // insertから除外
    public string InternalTag { get; set; }
}

var options = new InsertOptions { QueryPlacement = InsertQueryPlacement.Url };
await client.InsertBinaryAsync("events", columns, rows, options);

using ClickHouse.Driver.ADO.Parameters;

var parameters = new ClickHouseParameterCollection();
parameters.AddParameter("max_id", 100L);

var reader = await client.ExecuteReaderAsync(
    "SELECT * FROM default.my_table WHERE id < {max_id:Int64}",
    parameters
);

while (reader.Read())
{
    Console.WriteLine($"Id: {reader.GetInt64(0)}, Name: {reader.GetString(1)}");
}

// Define a POCO matching your result columns
public class SensorReading
{
    public ulong Id { get; set; }
    public DateTime Timestamp { get; set; }

    [ClickHouseColumn(Name = "sensor_name")]
    public string SensorName { get; set; }
    public double Value { get; set; }

}

// Register the type (once per client lifetime)
client.RegisterPocoType<SensorReading>();

// Stream results as typed objects
await foreach (var reading in client.QueryAsync<SensorReading>(
    "SELECT Id, sensor_name, Value, Timestamp FROM sensors"))
{
    Console.WriteLine($"{reading.SensorName}: {reading.Value}");
}

var reader = await client.ExecuteReaderAsync("SELECT Id, SensorName, Value, Timestamp FROM sensors");

while (reader.Read())
{
    SensorReading reading = reader.MapTo<SensorReading>();
    Console.WriteLine($"{reading.SensorName}: {reading.Value}");
}

SELECT {value:Array(UInt16)} as a

SELECT * FROM table WHERE val = {tuple_in_tuple:Tuple(UInt8, Tuple(String, UInt8))}

INSERT INTO table VALUES ({val1:Int32}, {val2:Array(UInt8)})

var parameters = new ClickHouseParameterCollection();
parameters.AddParameter("name", "my_database");

await client.ExecuteNonQueryAsync("CREATE DATABASE {name:Identifier}", parameters);

var parameters = new ClickHouseParameterCollection();
parameters.AddParameter("col", "user_id");

var reader = await client.ExecuteReaderAsync("SELECT {col:Identifier} FROM t", parameters);

var options = new QueryOptions
{
    QueryId = $"report-{Guid.NewGuid()}"
};

var reader = await client.ExecuteReaderAsync(
    "SELECT * FROM large_table",
    parameters: null,
    options: options
);

using ClickHouse.Driver.ADO.Parameters;

var settings = new ClickHouseClientSettings("Host=localhost")
{
    ParameterTypeResolver = new DictionaryParameterTypeResolver(new Dictionary<Type, string>
    {
        [typeof(DateTime)] = "DateTime64(3)",
        [typeof(decimal)] = "Decimal64(4)",
    }),
};
using var client = new ClickHouseClient(settings);

var parameters = new ClickHouseParameterCollection();
parameters.AddParameter("dt", DateTime.UtcNow);     // Mapped to DateTime64(3)
parameters.AddParameter("amount", 99.1234m);         // Mapped to Decimal64(4)

await client.ExecuteReaderAsync("SELECT @dt, @amount", parameters);

public class SmartDecimalResolver : IParameterTypeResolver
{
    public string ResolveType(Type clrType, object value, string parameterName)
    {
        if (clrType != typeof(decimal))
            return null; // Fall through to default

        var scale = (decimal.GetBits((decimal)value)[3] >> 16) & 0x7F;
        return scale <= 4 ? $"Decimal64({scale})" : $"Decimal128({scale})";
    }
}

using ClickHouse.Driver.ADO.Parameters;

var settings = new ClickHouseClientSettings("Host=localhost")
{
    ParameterFormatter = new DictionaryParameterFormatter(new Dictionary<Type, Func<object, string>>
    {
        [typeof(DateTime)] = v => ((DateTime)v).ToString("yyyy-MM-ddTHH:mm:ss.ffffff",
            System.Globalization.CultureInfo.InvariantCulture),
        [typeof(decimal)] = v => ((decimal)v).ToString("F4",
            System.Globalization.CultureInfo.InvariantCulture),
    }),
};
using var client = new ClickHouseClient(settings);

public class FixedDecimalFormatter : IParameterFormatter
{
    public string Format(object value, string typeName, string parameterName)
    {
        if (value is decimal d)
            return d.ToString("F4", System.Globalization.CultureInfo.InvariantCulture);
        return null; // Fall through for anything else
    }
}

using ClickHouse.Driver.ADO.Readers;

var converter = new DictionaryReadValueConverter()
    .For<DateTime>(dt => DateTime.SpecifyKind(dt, DateTimeKind.Utc))
    .For<string>(s => s.Trim());

var settings = new ClickHouseClientSettings("Host=localhost")
{
    ReadValueConverter = converter,
};
using var client = new ClickHouseClient(settings);

public class UtcKindForNoTzDateTimeConverter : IReadValueConverter
{
    public object ConvertValue(object value, string columnName, string clickHouseType)
    {
        if (value is DateTime dt && clickHouseType == "DateTime")
            return DateTime.SpecifyKind(dt, DateTimeKind.Utc);
        return value;
    }

    public T ConvertValue<T>(T value, string columnName, string clickHouseType)
    {
        if (typeof(T) == typeof(DateTime) && value is DateTime dt && clickHouseType == "DateTime")
            return (T)(object)DateTime.SpecifyKind(dt, DateTimeKind.Utc);
        return value;
    }
}

using var result = await client.ExecuteRawResultAsync(
    "SELECT * FROM default.my_table LIMIT 100 FORMAT JSONEachRow"
);

await using var stream = await result.ReadAsStreamAsync();
using var reader = new StreamReader(stream);
var json = await reader.ReadToEndAsync();

using var result = await client.ExecuteRawResultAsync(
    "SELECT * FROM events FORMAT Parquet",
    options: new QueryOptions { AcceptEncoding = "zstd" });

// Decode yourself or write to a file
await using var body = await result.ReadAsStreamAsync();

using var client = new ClickHouseClient(new ClickHouseClientSettings("Host=localhost")
{
    AcceptEncoding = "br",      // decodable, but not advertised by default
});

using var reader = await client.ExecuteReaderAsync(
    "SELECT * FROM events",
    options: new QueryOptions { AcceptEncoding = "identity" });   // opt this query out

Host=localhost;AcceptEncoding=br, gzip

using var reader = await client.ExecuteReaderAsync(
    "SELECT * FROM events",
    options: new QueryOptions { AcceptEncoding = "lz4" });   // decode this one with lz4 instead

using var result = await client.ExecuteRawResultAsync(
    "SELECT * FROM events FORMAT JSONEachRow",
    options: new QueryOptions { AcceptEncoding = "lz4" });

Console.WriteLine(result.ContentEncoding); // "lz4"

await using var body = await result.ReadDecompressedStreamAsync();
using var bodyReader = new StreamReader(body);
var json = await bodyReader.ReadToEndAsync();

var options = new InsertOptions { Compressor = GZipCompressor.Default };  // Content-Encoding: gzip
await client.InsertBinaryAsync("events", columns, rows, options);

var options = new InsertOptions { Compressor = new ZstdCompressor(level: 1) };

public sealed class MyCompressor : IClickHouseCompressor
{
    public string ContentEncoding => "my-codec";

    public Stream Compress(Stream destination, bool leaveOpen) => /* a compressing write stream */;
}

using var reader = await client.ExecuteReaderAsync(
    "SELECT * FROM events",
    options: new QueryOptions
    {
        AcceptEncoding = "zstd",
        CustomSettings = new Dictionary<string, object> { ["http_zlib_compression_level"] = 1 },
    });

foreach (var codec in new[] { "identity", "lz4", "zstd" })
{
    var sw = Stopwatch.StartNew();
    using var reader = await client.ExecuteReaderAsync(
        "SELECT ... FROM big_table",
        options: new QueryOptions { AcceptEncoding = codec });
    while (await reader.ReadAsync()) { }
    Console.WriteLine($"{codec,-9} {sw.ElapsedMilliseconds} ms");
}

SELECT ProfileEvents['UserTimeMicroseconds'] + ProfileEvents['SystemTimeMicroseconds'] AS cpu_us,
       ProfileEvents['NetworkSendBytes'] AS sent_bytes,
       query_duration_ms
FROM system.query_log
WHERE query_id = 'your-query-id' AND type = 'QueryFinish';

using var response = await client.InsertRawStreamAsync(
    table: "my_table",
    stream: File.OpenRead("data.csv"),
    format: "CSV",
    columns: ["id", "product", "price"] // Optional: specify columns
);

using ClickHouse.Driver.ADO;

// DataSourceは一度だけ作成する（DIにシングルトンとして登録する）
var dataSource = new ClickHouseDataSource("Host=localhost;Username=default;Password=secret");

// 必要に応じて軽量な接続を作成する
await using var connection = await dataSource.OpenConnectionAsync();

// 接続を使用する
await using var command = connection.CreateCommand("SELECT version()");
var version = await command.ExecuteScalarAsync();

// Startup.cs または Program.cs 内
services.AddSingleton(sp =>
{
    var factory = sp.GetRequiredService<IHttpClientFactory>();
    return new ClickHouseDataSource("Host=localhost", factory, "ClickHouse");
});

// サービス内
public class MyService
{
    private readonly ClickHouseDataSource _dataSource;

    public MyService(ClickHouseDataSource dataSource)
    {
        _dataSource = dataSource;
    }

    public async Task DoWorkAsync()
    {
        await using var connection = await _dataSource.OpenConnectionAsync();
        // 接続を使用...
    }
}

await using var connection = await dataSource.OpenConnectionAsync();

// SQLでコマンドを作成する
await using var command = connection.CreateCommand("SELECT * FROM my_table WHERE id = {id:Int64}");
command.AddParameter("id", 42L);

// 実行して結果を読み取る
await using var reader = await command.ExecuteReaderAsync();
while (reader.Read())
{
    Console.WriteLine($"Name: {reader.GetString("name")}");
}

await using var reader = await command.ExecuteReaderAsync();

while (reader.Read())
{
    // カラムインデックスによるアクセス
    var id = reader.GetInt64(0);
    var name = reader.GetString(1);

    // カラム名によるアクセス
    var email = reader.GetString("email");

    // 汎用アクセス
    var timestamp = reader.GetFieldValue<DateTime>("created_at");

    // nullチェック
    if (!reader.IsDBNull("optional_field"))
    {
        var value = reader.GetString("optional_field");
    }
}

if (reader.TryGetEnumOrdinal(ordinal, out int value))
    Console.WriteLine(value);   // e.g. 1 for 'Active' in Enum8('Active' = 1)

// CustomSettingsを使用する場合
var settings = new ClickHouseClientSettings("Host=localhost");
settings.CustomSettings["async_insert"] = 1;
settings.CustomSettings["wait_for_async_insert"] = 1; // 推奨: フラッシュの完了確認を待機する

// または接続文字列を使用する場合
// "Host=localhost;set_async_insert=1;set_wait_for_async_insert=1"

var settings = new ClickHouseClientSettings
{
    Host = "localhost",
    UseSession = true,
    SessionId = "my-session", // Optional -- will be auto-generated if not provided
};

using var client = new ClickHouseClient(settings);

await client.ExecuteNonQueryAsync("CREATE TEMPORARY TABLE temp_ids (id UInt64)");
await client.ExecuteNonQueryAsync("INSERT INTO temp_ids VALUES (1), (2), (3)");

var reader = await client.ExecuteReaderAsync(
    "SELECT * FROM users WHERE id IN (SELECT id FROM temp_ids)"
);

var settings = new ClickHouseClientSettings
{
    Host = "localhost",
    UseSession = true,
    SessionId = "my-session",
};

var dataSource = new ClickHouseDataSource(settings);
await using var connection = await dataSource.OpenConnectionAsync();

await using var cmd1 = connection.CreateCommand("CREATE TEMPORARY TABLE temp_ids (id UInt64)");
await cmd1.ExecuteNonQueryAsync();

await using var cmd2 = connection.CreateCommand("INSERT INTO temp_ids VALUES (1), (2), (3)");
await cmd2.ExecuteNonQueryAsync();

await using var cmd3 = connection.CreateCommand("SELECT * FROM users WHERE id IN (SELECT id FROM temp_ids)");
await using var reader = await cmd3.ExecuteReaderAsync();

var reader = (ClickHouseDataReader)await connection.ExecuteReaderAsync(
    "SELECT toDateTime('2024-06-15 14:30:00', 'Europe/Amsterdam')");
reader.Read();

var dt = reader.GetDateTime(0);    // 2024-06-15 14:30:00, Kind=Unspecified（未指定）
var dto = reader.GetDateTimeOffset(0); // 2024-06-15 14:30:00 +02:00 (CEST)

// Configure string mode via settings
var settings = new ClickHouseClientSettings("Host=localhost")
{
    JsonReadMode = JsonReadMode.String
};

// Or via connection string
// "Host=localhost;JsonReadMode=String"

// Column type JSON(x Nullable(Int64))
// stored '{"x":null}'  ->  {"x":null}
// stored '{}'          ->  {"x":null}

var settings = new ClickHouseClientSettings("Host=localhost")
{
    AllowDuplicateJsonKeys = true
};

// Or via connection string
// "Host=localhost;AllowDuplicateJsonKeys=true"

// Configure key-value-pair mode via settings
var settings = new ClickHouseClientSettings("Host=localhost")
{
    MapReadMode = MapReadMode.KeyValuePairs
};

// Or via connection string
// "Host=localhost;MapReadMode=KeyValuePairs"

var utcTime = new DateTime(2024, 1, 15, 12, 0, 0, DateTimeKind.Utc);
// Stored as 12:00 UTC
// Read from DateTime('Europe/Amsterdam') column: 13:00 (UTC+1)
// Read from DateTime('UTC') column: 12:00 UTC

var wallClock = new DateTime(2024, 1, 15, 14, 30, 0, DateTimeKind.Unspecified);
// Written to DateTime('Europe/Amsterdam') column: stored as 14:30 Amsterdam time
// Read back from DateTime('Europe/Amsterdam') column: 14:30

// 正しい例: SQLの型ヒントにタイムゾーンを指定 - 型は自動的に抽出される
command.CommandText = "INSERT INTO table (dt_amsterdam) VALUES ({dt:DateTime('Europe/Amsterdam')})";
command.AddParameter("dt", myDateTime);

// 誤った例: タイムゾーンヒントなしの場合、UTCとして解釈される
command.CommandText = "INSERT INTO table (dt_amsterdam) VALUES ({dt:DateTime})";
command.AddParameter("dt", myDateTime);
// 文字列値 "2024-01-15 14:30:00" はアムステルダム時間ではなくUTCとして解釈される！