Repository navigation
Expand file tree
/
Copy pathInsert_002_BulkInsert.cs
More file actions
168 lines (143 loc) · 7.45 KB
/
Copy pathInsert_002_BulkInsert.cs
File metadata and controls
168 lines (143 loc) · 7.45 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
using ClickHouse.Driver.Compression;
using ClickHouse.Driver.Utility;
namespace ClickHouse.Driver.Examples;
/// <summary>
/// Demonstrates high-performance bulk data insertion using InsertBinaryAsync.
/// This is the recommended approach for inserting large amounts of data efficiently.
/// </summary>
public static class BulkInsert
{
public static async Task Run()
{
using var client = ExampleConfig.CreateHttpClient();
var tableName = "example_bulk_insert";
// Create a test table
await client.ExecuteNonQueryAsync($@"
CREATE TABLE IF NOT EXISTS {tableName}
(
id UInt64,
product_name String,
category String,
price Float32,
quantity UInt32,
sale_date DateTime
)
ENGINE = MergeTree()
ORDER BY (id)
");
Console.WriteLine($"Table '{tableName}' created\n");
// Example 1: Bulk insert with default options
Console.WriteLine("1. Bulk inserting with InsertBinaryAsync:");
var columns = new[] { "id", "product_name", "category", "price", "quantity", "sale_date" };
var data = GenerateSampleData(10000, startId: 1);
var rowsInserted = await client.InsertBinaryAsync(tableName, columns, data);
Console.WriteLine($" Inserted {rowsInserted} rows\n");
// Example 2: Bulk insert with custom options (batch size, parallelism)
Console.WriteLine("2. Bulk inserting with custom InsertOptions:");
var options = new InsertOptions
{
BatchSize = 5000, // Rows per batch
MaxDegreeOfParallelism = 4, // Parallel batch uploads
};
var moreData = GenerateSampleData(10000, startId: 10001);
rowsInserted = await client.InsertBinaryAsync(tableName, columns, moreData, options);
Console.WriteLine($" Inserted {rowsInserted} rows with custom options\n");
// Example 3: Bulk insert with specific columns (others use defaults)
Console.WriteLine("3. Bulk inserting with specific columns:");
var partialTableName = "example_bulk_partial";
await client.ExecuteNonQueryAsync($@"
CREATE TABLE IF NOT EXISTS {partialTableName}
(
id UInt64,
name String,
value Float32 DEFAULT 0.0,
created_at DateTime DEFAULT now()
)
ENGINE = MergeTree()
ORDER BY (id)
");
var partialColumns = new[] { "id", "name" };
var partialData = new List<object[]>
{
new object[] { 1UL, "Item 1" },
new object[] { 2UL, "Item 2" },
new object[] { 3UL, "Item 3" },
};
rowsInserted = await client.InsertBinaryAsync(partialTableName, partialColumns, partialData);
Console.WriteLine($" Inserted {rowsInserted} rows with partial columns\n");
// Example 4: Bulk insert with Brotli compression
Console.WriteLine("4. Bulk inserting with Brotli compression:");
// The insert body is compressed before being sent. The default is ZSTD at level 3.
// Set the Compressor property to switch codecs; the built-in GZipCompressor and
// BrotliCompressor also expose a CompressionLevel. Behind a proxy tier that re-encodes request
// bodies, GZip is the safest choice: a request's Content-Encoding is a declaration rather than
// an offer, so an intermediary that does not understand the codec fails the insert outright.
//
// Over a fast/local link, set Compressor = null to skip compression entirely;
// there the compression cost outweighs the bandwidth saved.
var brotliOptions = new InsertOptions
{
Compressor = BrotliCompressor.Default,
};
var brotliData = GenerateSampleData(10000, startId: 20001);
rowsInserted = await client.InsertBinaryAsync(tableName, columns, brotliData, brotliOptions);
Console.WriteLine($" Inserted {rowsInserted} rows using Brotli compression\n");
// Example 5: Bulk insert with LZ4 compression at a custom level
Console.WriteLine("5. Bulk inserting with LZ4 compression (custom level):");
// LZ4 ships in the core driver (no extra package needed). Lz4Compressor.Default uses fast mode
// (Lz4Level.Fast), the recommended setting for almost all inserts — it is fastest and puts the
// least decompression load on the server. Higher levels cost noticeably more CPU for
// little-to-no extra compression on typical data.
var lz4Options = new InsertOptions
{
Compressor = new Lz4Compressor(Lz4Level.Fast),
};
var lz4Data = GenerateSampleData(10000, startId: 30001);
rowsInserted = await client.InsertBinaryAsync(tableName, columns, lz4Data, lz4Options);
Console.WriteLine($" Inserted {rowsInserted} rows using LZ4 compression (L03_HC)\n");
// Query and display sample results
Console.WriteLine("Sample data from main table:");
using (var reader = await client.ExecuteReaderAsync($"SELECT * FROM {tableName} ORDER BY id LIMIT 5"))
{
Console.WriteLine("ID\tProduct Name\t\tCategory\tPrice\t\tQuantity");
Console.WriteLine("--\t------------\t\t--------\t-----\t\t--------");
while (reader.Read())
{
var id = reader.GetFieldValue<ulong>(0);
var productName = reader.GetString(1);
var category = reader.GetString(2);
var price = reader.GetFloat(3);
var quantity = reader.GetFieldValue<uint>(4);
Console.WriteLine($"{id}\t{productName,-20}\t{category,-15}\t${price,-10:F2}\t{quantity}");
}
}
// Get total row counts
var totalCount = await client.ExecuteScalarAsync($"SELECT count() FROM {tableName}");
Console.WriteLine($"\nTotal rows in {tableName}: {totalCount}");
var partialCount = await client.ExecuteScalarAsync($"SELECT count() FROM {partialTableName}");
Console.WriteLine($"Total rows in {partialTableName}: {partialCount}");
// Clean up
await client.ExecuteNonQueryAsync($"DROP TABLE IF EXISTS {tableName}");
await client.ExecuteNonQueryAsync($"DROP TABLE IF EXISTS {partialTableName}");
Console.WriteLine($"\nTables dropped");
}
private static IEnumerable<object[]> GenerateSampleData(int count, ulong startId = 1)
{
var random = new Random(42);
var categories = new[] { "Electronics", "Furniture", "Clothing", "Books", "Toys" };
var productPrefixes = new[] { "Premium", "Deluxe", "Standard", "Economy", "Budget" };
var productTypes = new[] { "Widget", "Gadget", "Device", "Tool", "Item" };
for (ulong i = 0; i < (ulong)count; i++)
{
var id = startId + i;
var prefix = productPrefixes[random.Next(productPrefixes.Length)];
var type = productTypes[random.Next(productTypes.Length)];
var productName = $"{prefix} {type} #{id}";
var category = categories[random.Next(categories.Length)];
var price = (float)(random.NextDouble() * 1000 + 10);
var quantity = (uint)random.Next(1, 100);
var saleDate = DateTime.UtcNow.AddDays(-random.Next(0, 365));
yield return new object[] { id, productName, category, price, quantity, saleDate };
}
}
}