Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions ClickHouse.Driver.Tests/ADO/ConnectionTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -523,6 +523,21 @@ public async Task ShouldUseQueryIdForRawStream()
Assert.That(queryResult.QueryId, Is.EqualTo(queryId));
}

[Test]
public async Task PostStreamAsync_AfterChangeDatabase_RunsInSelectedDatabase([Values] bool useCallback, [Values] bool withQueryId)
{
using var conn = client.CreateConnection();
conn.ChangeDatabase("system");
const string sql = "SELECT currentDatabase() FORMAT TabSeparated";
var queryId = withQueryId ? Guid.NewGuid().ToString() : null;

using var response = useCallback
? await conn.PostStreamAsync(sql, (_, _) => Task.CompletedTask, false, CancellationToken.None, queryId)
: await conn.PostStreamAsync(sql, new MemoryStream(), false, CancellationToken.None, queryId);

Assert.That(await response.Content.ReadAsStringAsync(), Is.EqualTo("system\n"));
}

private static string[] GetColumnNames(DataTable table) => table.Columns.Cast<DataColumn>().Select(dc => dc.ColumnName).ToArray();

[Test]
Expand Down
18 changes: 18 additions & 0 deletions ClickHouse.Driver.Tests/BulkCopy/BulkCopyTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,24 @@ public async Task ShouldInsertStringOrFixedString(string columnType, string tabl
Assert.That(data, Is.EqualTo(testData));
}

[Test]
[TestCase(RowBinaryFormat.RowBinary)]
[TestCase(RowBinaryFormat.RowBinaryWithDefaults)]
public async Task WriteToServerAsync_AfterChangeDatabase_InsertsIntoSelectedDatabase(RowBinaryFormat format)
{
await client.ExecuteNonQueryAsync("CREATE DATABASE IF NOT EXISTS test_secondary");
var targetTable = CreateTableName($"bulk_{format}", database: "test_secondary");
await connection.ExecuteStatementAsync($"CREATE TABLE {targetTable} (value UInt32) ENGINE Memory");

using var conn = client.CreateConnection();
using var bulkCopy = new ClickHouseBulkCopy(conn, format) { DestinationTableName = TestUtilities.BareTableName(targetTable) };
conn.ChangeDatabase("test_secondary");

await bulkCopy.WriteToServerAsync(new[] { new object[] { 1u }, new object[] { 2u } });

Assert.That(await client.ExecuteScalarAsync($"SELECT count() FROM {targetTable}"), Is.EqualTo(2UL));
}

[Test]
public async Task WriteToServerAsync_FixedStringWithNonSeekableStreamTooLong_ThrowsArgumentException()
{
Expand Down
19 changes: 6 additions & 13 deletions ClickHouse.Driver/ADO/ClickHouseConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,7 @@ public override Task OpenAsync(CancellationToken cancellationToken)
/// <returns>Task-wrapped HttpResponseMessage object</returns>
public async Task<HttpResponseMessage> PostStreamAsync(string sql, Stream data, bool isCompressed, CancellationToken token, string queryId = null)
{
var options = GetQueryOptionsWithQueryId(queryId);
var options = GetPostStreamOptions(queryId);
return await ClickHouseClient.PostStreamAsync(sql, data, isCompressed, token, options).ConfigureAwait(false);
}

Expand All @@ -247,22 +247,15 @@ public async Task<HttpResponseMessage> PostStreamAsync(string sql, Stream data,
/// <returns>Task-wrapped HttpResponseMessage object</returns>
public async Task<HttpResponseMessage> PostStreamAsync(string sql, Func<Stream, CancellationToken, Task> callback, bool isCompressed, CancellationToken token, string queryId = null)
{
var options = GetQueryOptionsWithQueryId(queryId);
var options = GetPostStreamOptions(queryId);
return await ClickHouseClient.PostStreamAsync(sql, callback, isCompressed, token, options).ConfigureAwait(false);
}

private static QueryOptions GetQueryOptionsWithQueryId(string queryId)
private QueryOptions GetPostStreamOptions(string queryId) => new()
{
QueryOptions options = null;
if (queryId != null)
{
options = new QueryOptions
{
QueryId = queryId,
};
}
return options;
}
QueryId = queryId,
Database = Database,
Comment thread
polyglotAI-bot marked this conversation as resolved.
Resolved
};

#pragma warning disable CS0109 // Member does not hide an inherited member; new keyword is not required
public new ClickHouseCommand CreateCommand(string commandText = null) => new (this) { CommandText = commandText };
Expand Down
1 change: 1 addition & 0 deletions ClickHouse.Driver/Copy/ClickHouseBulkCopy.cs
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,7 @@ public async Task WriteToServerAsync(IEnumerable<object[]> rows, CancellationTok
{
var options = new InsertOptions
{
Database = connection.Database,
BatchSize = BatchSize,
MaxDegreeOfParallelism = MaxDegreeOfParallelism,
Format = rowBinaryFormat,
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
* Fixed `ClickHouseConnection.PostStreamAsync` and `ClickHouseBulkCopy` ignoring the database
selected with `ChangeDatabase` ([#683](https://github.com/ClickHouse/clickhouse-cs/issues/683)).
An unqualified table name now resolves in that database instead of the configured one.
23 changes: 23 additions & 0 deletions docs/http.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -1430,6 +1430,29 @@ any column that is not an enum. The ordinal is the
signed value from the wire, so it can be negative, and an `Enum16` ordinal can be larger than a
byte.

---

### Changing the database {#ado-net-change-database}

`ChangeDatabase` selects the database for one connection. The connection sends this database with
each request, so an unqualified table name resolves in it:

```csharp
// connection: a ClickHouseConnection opened from a ClickHouseDataSource
connection.ChangeDatabase("analytics");

await using var command = connection.CreateCommand("SELECT count() FROM events"); // analytics.events
var count = await command.ExecuteScalarAsync();
```

- Commands that run on the connection and both `PostStreamAsync` overloads of the connection use
the selected database.
- `ClickHouseBulkCopy` reads the database when `WriteToServerAsync` runs, so a `ChangeDatabase` call
made after you create the bulk copy applies to the next write.
- `connection.Database` returns the selected database. Before the first `ChangeDatabase` call, it
returns the `Database` setting.
- `ChangeDatabase` does not change the settings of the `ClickHouseClient` or of other connections.

## Best practices {#best-practices}

### Connection lifetime and pooling {#best-practices-connection-lifetime}
Expand Down
Loading