Google Git
Sign in
apache/iotdb-client-csharp/refs/heads/codex/add-mtls-support/./src/Apache.IoTDB/SessionPool.cs
blob: 5392859ae649bf65daad936d18c5d082fbc6d663 [file]
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Security;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Apache.IoTDB.DataStructure;
using Microsoft.Extensions.Logging;
using Thrift;
using Thrift.Protocol;
using Thrift.Transport;
using Thrift.Transport.Client;
namespace Apache.IoTDB
{
public partial class SessionPool : IDisposable, IPoolDiagnosticReporter
{
private static readonly TSProtocolVersion ProtocolVersion = TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3;
private const string DepletionReasonReconnectFailed = "Reconnection failed";
private readonly string _username;
private readonly string _password;
private bool _enableRpcCompression;
private string _zoneId;
private readonly List<string> _nodeUrls = new();
private readonly List<TEndPoint> _endPoints = new();
private readonly string _host;
private readonly int _port;
private readonly bool _useSsl;
private readonly string _clientCertificatePath;
private readonly string _clientCertificatePassword;
private readonly string _rootCertificatePath;
private X509Certificate2 _clientCertificate;
private X509Certificate2Collection _rootCertificates;
private RemoteCertificateValidationCallback _remoteCertificateValidationCallback;
private LocalCertificateSelectionCallback _localCertificateSelectionCallback;
private readonly int _fetchSize;
/// <summary>
/// _timeout is the amount of time a Session will wait for a send operation to complete successfully.
/// </summary>
private readonly int _timeout;
private readonly int _poolSize = 4;
private readonly string _sqlDialect = IoTDBConstant.TREE_SQL_DIALECT;
private string _database;
private readonly Utils _utilFunctions = new();
private const int RetryNum = 3;
private const string ServerAuthenticationOid = "1.3.6.1.5.5.7.3.1";
private readonly object _tlsConfigurationLock = new();
private bool _tlsConfigurationLoaded;
private bool _debugMode;
private bool _isClose = true;
private ConcurrentClientQueue _clients;
private ILogger _logger;
private PoolHealthMetrics _healthMetrics;
public delegate Task<TResult> AsyncOperation<TResult>(Client client);
/// <summary>
/// Retrieves current count of idle clients ready for operations.
/// </summary>
public int AvailableClients => _clients?.ClientQueue.Count ?? 0;
/// <summary>
/// Retrieves the configured maximum capacity of the session pool.
/// </summary>
public int TotalPoolSize => _healthMetrics?.GetConfiguredMaxSize() ?? _poolSize;
/// <summary>
/// Retrieves cumulative tally of reconnection failures since pool was opened.
/// </summary>
public int FailedReconnections => _healthMetrics?.GetReconnectionFailureTally() ?? 0;
[Obsolete("This method is deprecated, please use new SessionPool.Builder().")]
public SessionPool(string host, int port, int poolSize)
: this(host, port, "root", "root", 1024, "Asia/Shanghai", poolSize, true, 60)
{
}
[Obsolete(" This method is deprecated, please use new SessionPool.Builder().")]
public SessionPool(string host, int port, string username, string password)
: this(host, port, username, password, 1024, "Asia/Shanghai", 8, true, 60)
{
}
public SessionPool(string host, int port, string username, string password, int fetchSize)
: this(host, port, username, password, fetchSize, "Asia/Shanghai", 8, true, 60)
{
}
public SessionPool(string host, int port) : this(host, port, "root", "root", 1024, "Asia/Shanghai", 8, true, 60)
{
}
public SessionPool(string host, int port, string username, string password, int fetchSize, string zoneId, int poolSize, bool enableRpcCompression, int timeout)
: this(host, port, username, password, fetchSize, zoneId, poolSize, enableRpcCompression, timeout, false, null, null, null, IoTDBConstant.TREE_SQL_DIALECT, "")
{
}
protected internal SessionPool(string host, int port, string username, string password, int fetchSize, string zoneId, int poolSize, bool enableRpcCompression, int timeout, bool useSsl, string clientCertificatePath, string clientCertificatePassword, string rootCertificatePath, string sqlDialect, string database)
{
_host = host;
_port = port;
_username = username;
_password = password;
_zoneId = zoneId;
_fetchSize = fetchSize;
_debugMode = false;
_poolSize = poolSize;
_enableRpcCompression = enableRpcCompression;
_timeout = timeout;
_useSsl = useSsl;
_clientCertificatePath = clientCertificatePath;
_clientCertificatePassword = clientCertificatePassword;
_rootCertificatePath = rootCertificatePath;
_sqlDialect = sqlDialect;
_database = database;
}
/// <summary>
/// Initializes a new instance of the <see cref="SessionPool"/> class.
/// </summary>
/// <param name="nodeUrls">The list of node URLs to connect to, multiple ip:rpcPort eg.127.0.0.1:9001</param>
/// <param name="poolSize">The size of the session pool.</param>
public SessionPool(List<string> nodeUrls, int poolSize)
: this(nodeUrls, "root", "root", 1024, "Asia/Shanghai", poolSize, true, 60)
{
}
public SessionPool(List<string> nodeUrls, string username, string password)
: this(nodeUrls, username, password, 1024, "Asia/Shanghai", 8, true, 60)
{
}
public SessionPool(List<string> nodeUrls, string username, string password, int fetchSize)
: this(nodeUrls, username, password, fetchSize, "Asia/Shanghai", 8, true, 60)
{
}
public SessionPool(List<string> nodeUrls, string username, string password, int fetchSize, string zoneId)
: this(nodeUrls, username, password, fetchSize, zoneId, 8, true, 60)
{
}
public SessionPool(List<string> nodeUrls, string username, string password, int fetchSize, string zoneId, int poolSize, bool enableRpcCompression, int timeout)
: this(nodeUrls, username, password, fetchSize, zoneId, poolSize, enableRpcCompression, timeout, false, null, null, null, IoTDBConstant.TREE_SQL_DIALECT, "")
{
}
protected internal SessionPool(List<string> nodeUrls, string username, string password, int fetchSize, string zoneId, int poolSize, bool enableRpcCompression, int timeout, bool useSsl, string clientCertificatePath, string clientCertificatePassword, string rootCertificatePath, string sqlDialect, string database)
{
if (nodeUrls.Count == 0)
{
throw new ArgumentException("nodeUrls shouldn't be empty.");
}
_nodeUrls = nodeUrls;
_endPoints = _utilFunctions.ParseSeedNodeUrls(nodeUrls);
_username = username;
_password = password;
_zoneId = zoneId;
_fetchSize = fetchSize;
_debugMode = false;
_poolSize = poolSize;
_enableRpcCompression = enableRpcCompression;
_timeout = timeout;
_useSsl = useSsl;
_clientCertificatePath = clientCertificatePath;
_clientCertificatePassword = clientCertificatePassword;
_rootCertificatePath = rootCertificatePath;
_sqlDialect = sqlDialect;
_database = database;
}
public async Task<TResult> ExecuteClientOperationAsync<TResult>(AsyncOperation<TResult> operation, string errMsg, bool retryOnFailure = true, bool putClientBack = true)
{
Client client = _clients.Take();
bool shouldReturnClient = true;
bool operationSucceeded = false;
try
{
var resp = await operation(client);
operationSucceeded = true;
return resp;
}
catch (Exception ex)
{
if (retryOnFailure)
{
// Try to reconnect
try
{
client = await Reconnect(client);
// Reconnect succeeded, client is now a new healthy connection
}
catch (ReconnectionFailedException reconnectEx)
{
// Reconnection failed - original client was closed by Reconnect
shouldReturnClient = false;
throw new SessionPoolDepletedException(DepletionReasonReconnectFailed, AvailableClients, TotalPoolSize, FailedReconnections, reconnectEx);
}
// Reconnect succeeded, try the operation again
try
{
var resp = await operation(client);
operationSucceeded = true;
return resp;
}
catch (Exception retryEx)
{
// Retry operation failed, but client is healthy and should be returned to pool
// shouldReturnClient remains true
string detailedMsg = $"{errMsg}. {retryEx.Message}";
throw new TException(detailedMsg, retryEx);
}
}
else
{
// Preserve original error message from server
string detailedMsg = $"{errMsg}. {ex.Message}";
throw new TException(detailedMsg, ex);
}
}
finally
{
// Return client to pool if:
// 1. putClientBack is true (normal operations - client should always be returned), OR
// 2. putClientBack is false (query operations) BUT operation failed, meaning SessionDataSet
// wasn't created and won't manage the client
// Do NOT return if reconnection failed (shouldReturnClient is false) because client was closed by Reconnect
bool shouldReturnForQueryFailure = !putClientBack && !operationSucceeded;
if (shouldReturnClient && (putClientBack || shouldReturnForQueryFailure))
{
_clients.Add(client);
}
}
}
/// <summary>
/// Gets or sets the amount of time a Session will wait for a send operation to complete successfully.
/// </summary>
/// <remarks> The send time-out value, in milliseconds. The default is 10000.</remarks>
public int TimeOut { get; set; } = 10000;
ILoggerFactory factory;
private bool disposedValue;
public void OpenDebugMode(Action<ILoggingBuilder> configure)
{
_debugMode = true;
factory = LoggerFactory.Create(configure);
_logger = factory.CreateLogger(nameof(Apache.IoTDB));
}
public void CloseDebugMode()
{
_debugMode = false;
}
public async Task Open(bool enableRpcCompression, CancellationToken cancellationToken = default)
{
_enableRpcCompression = enableRpcCompression;
await Open(cancellationToken);
}
public async Task Open(CancellationToken cancellationToken = default)
{
_healthMetrics = new PoolHealthMetrics(_poolSize);
_clients = new ConcurrentClientQueue();
_clients.Timeout = _timeout * 5;
_clients.DiagnosticReporter = this;
if (_nodeUrls.Count == 0)
{
for (var index = 0; index < _poolSize; index++)
{
try
{
_clients.Add(await CreateAndOpen(_host, _port, _enableRpcCompression, _timeout, _useSsl, _sqlDialect, _database, cancellationToken));
}
catch (Exception e)
{
_logger?.LogWarning(e, "Failed to create connection {0}/{1} to {2}:{3}", index + 1, _poolSize, _host, _port);
}
}
}
else
{
int startIndex = 0;
for (var index = 0; index < _poolSize; index++)
{
bool isConnected = false;
for (int i = 0; i < _endPoints.Count; i++)
{
var endPointIndex = (startIndex + i) % _endPoints.Count;
var endPoint = _endPoints[endPointIndex];
try
{
var client = await CreateAndOpen(endPoint.Ip, endPoint.Port, _enableRpcCompression, _timeout, _useSsl, _sqlDialect, _database, cancellationToken);
_clients.Add(client);
isConnected = true;
startIndex = (endPointIndex + 1) % _endPoints.Count;
break;
}
catch (Exception e)
{
_logger?.LogWarning(e, "Failed to create connection to {0}:{1}", endPoint.Ip, endPoint.Port);
}
}
if (!isConnected) // current client could not connect to any endpoint
{
throw new TException("Error occurs when opening session pool. Could not connect to any server", null);
}
}
}
if (_clients.ClientQueue.Count != _poolSize)
{
throw new TException(string.Format("Error occurs when opening session pool. Client pool size is not equal to the expected size. Client pool size: {0}, expected size: {1}, Please check the server status", _clients.ClientQueue.Count, _poolSize), null);
}
_isClose = false;
}
public async Task<Client> Reconnect(Client originalClient = null, CancellationToken cancellationToken = default)
{
originalClient?.Transport.Close();
if (_nodeUrls.Count == 0)
{
for (int attempt = 1; attempt <= RetryNum; attempt++)
{
try
{
var client = await CreateAndOpen(_host, _port, _enableRpcCompression, _timeout, _useSsl, _sqlDialect, _database, cancellationToken);
return client;
}
catch (Exception e)
{
_logger?.LogWarning(e, "Reconnection attempt {0}/{1} to {2}:{3} failed", attempt, RetryNum, _host, _port);
}
}
}
else
{
int startIndex = _endPoints.FindIndex(x => x.Ip == originalClient.EndPoint.Ip && x.Port == originalClient.EndPoint.Port);
if (startIndex == -1)
{
throw new ArgumentException($"The original client is not in the list of endpoints. Original client: {originalClient.EndPoint.Ip}:{originalClient.EndPoint.Port}");
}
for (int attempt = 1; attempt <= RetryNum; attempt++)
{
for (int i = 0; i < _endPoints.Count; i++)
{
int j = (startIndex + i) % _endPoints.Count;
try
{
var client = await CreateAndOpen(_endPoints[j].Ip, _endPoints[j].Port, _enableRpcCompression, _timeout, _useSsl, _sqlDialect, _database, cancellationToken);
return client;
}
catch (Exception e)
{
_logger?.LogWarning(e, "Reconnection attempt {0}/{1} to {2}:{3} failed", attempt, RetryNum, _endPoints[j].Ip, _endPoints[j].Port);
}
}
}
}
_healthMetrics?.IncrementReconnectionFailures();
throw new ReconnectionFailedException("Error occurs when reconnecting session pool. Could not connect to any server");
}
public bool IsOpen() => !_isClose;
public async Task Close()
{
if (_isClose)
{
return;
}
foreach (var client in _clients.ClientQueue.AsEnumerable())
{
var closeSessionRequest = new TSCloseSessionReq(client.SessionId);
try
{
await client.ServiceClient.closeSessionAsync(closeSessionRequest);
}
catch (TException e)
{
throw new TException("Error occurs when closing session at server. Maybe server is down", e);
}
finally
{
_isClose = true;
client.Transport?.Close();
}
}
}
public async Task SetTimeZone(string zoneId)
{
_zoneId = zoneId;
foreach (var client in _clients.ClientQueue.AsEnumerable())
{
var req = new TSSetTimeZoneReq(client.SessionId, zoneId);
try
{
var resp = await client.ServiceClient.setTimeZoneAsync(req);
if (_debugMode)
{
_logger.LogInformation("setting time zone_id as {0}, server message:{1}", zoneId, resp.Message);
}
}
catch (TException e)
{
throw new TException("could not set time zone", e);
}
}
}
public async Task<string> GetTimeZone()
{
if (_zoneId != "")
{
return _zoneId;
}
var client = _clients.Take();
try
{
var response = await client.ServiceClient.getTimeZoneAsync(client.SessionId);
return response?.TimeZone;
}
catch (TException e)
{
throw new TException("could not get time zone", e);
}
finally
{
_clients.Add(client);
}
}
private async Task<Client> CreateAndOpen(string host, int port, bool enableRpcCompression, int timeout, bool useSsl, string sqlDialect, string database, CancellationToken cancellationToken = default)
{
TTransport socket;
if (useSsl)
{
EnsureTlsConfigurationLoaded();
socket = IPAddress.TryParse(host, out var ipAddress)
? new TTlsSocketTransport(ipAddress, port, null, timeout, _clientCertificate, _remoteCertificateValidationCallback, _localCertificateSelectionCallback)
: new TTlsSocketTransport(host, port, null, timeout, _clientCertificate, _remoteCertificateValidationCallback, _localCertificateSelectionCallback);
}
else
{
socket = new TSocketTransport(host, port, false, null, timeout);
}
var transport = new TFramedTransport(socket);
if (!transport.IsOpen)
{
await transport.OpenAsync(cancellationToken);
}
var client = enableRpcCompression ?
new IClientRPCService.Client(new TCompactProtocol(transport)) :
new IClientRPCService.Client(new TBinaryProtocol(transport));
var openReq = new TSOpenSessionReq(ProtocolVersion, _zoneId, _username)
{
Password = _password,
};
if (openReq.Configuration == null)
{
openReq.Configuration = new Dictionary<string, string>();
}
openReq.Configuration.Add("sql_dialect", sqlDialect);
if (!String.IsNullOrEmpty(database))
{
openReq.Configuration.Add("db", database);
}
try
{
var openResp = await client.openSessionAsync(openReq, cancellationToken);
if (openResp.ServerProtocolVersion != ProtocolVersion)
{
throw new TException($"Protocol Differ, Client version is {ProtocolVersion} but Server version is {openResp.ServerProtocolVersion}", null);
}
if (openResp.ServerProtocolVersion == 0)
{
throw new TException("Protocol not supported", null);
}
var sessionId = openResp.SessionId;
var statementId = await client.requestStatementIdAsync(sessionId, cancellationToken);
var endpoint = new TEndPoint(host, port);
var returnClient = new Client(
client,
sessionId,
statementId,
transport,
endpoint);
return returnClient;
}
catch (Exception)
{
transport.Close();
throw;
}
}
private void EnsureTlsConfigurationLoaded()
{
lock (_tlsConfigurationLock)
{
if (_tlsConfigurationLoaded)
{
return;
}
_clientCertificate = LoadClientCertificate(_clientCertificatePath, _clientCertificatePassword);
_rootCertificates = LoadRootCertificates(_rootCertificatePath);
_remoteCertificateValidationCallback = CreateRemoteCertificateValidationCallback(_rootCertificates);
_localCertificateSelectionCallback = CreateLocalCertificateSelectionCallback(_clientCertificate);
_tlsConfigurationLoaded = true;
}
}
private static X509Certificate2 LoadClientCertificate(string clientCertificatePath, string clientCertificatePassword)
{
if (string.IsNullOrWhiteSpace(clientCertificatePath))
{
return null;
}
return string.IsNullOrEmpty(clientCertificatePassword)
? new X509Certificate2(clientCertificatePath)
: new X509Certificate2(clientCertificatePath, clientCertificatePassword);
}
private static X509Certificate2Collection LoadRootCertificates(string rootCertificatePath)
{
if (string.IsNullOrWhiteSpace(rootCertificatePath))
{
return null;
}
var certificateBytes = File.ReadAllBytes(rootCertificatePath);
var certificates = LoadPemCertificates(certificateBytes);
if (certificates.Count > 0)
{
return certificates;
}
certificates.Import(certificateBytes);
return certificates;
}
private static X509Certificate2Collection LoadPemCertificates(byte[] certificateBytes)
{
const string beginCertificate = "-----BEGIN CERTIFICATE-----";
const string endCertificate = "-----END CERTIFICATE-----";
var certificates = new X509Certificate2Collection();
var certificateText = Encoding.ASCII.GetString(certificateBytes);
var startIndex = certificateText.IndexOf(beginCertificate, StringComparison.Ordinal);
while (startIndex >= 0)
{
startIndex += beginCertificate.Length;
var endIndex = certificateText.IndexOf(endCertificate, startIndex, StringComparison.Ordinal);
if (endIndex < 0)
{
break;
}
var base64 = certificateText.Substring(startIndex, endIndex - startIndex)
.Replace("\r", string.Empty)
.Replace("\n", string.Empty)
.Trim();
certificates.Add(new X509Certificate2(Convert.FromBase64String(base64)));
startIndex = certificateText.IndexOf(beginCertificate, endIndex + endCertificate.Length, StringComparison.Ordinal);
}
return certificates;
}
private static RemoteCertificateValidationCallback CreateRemoteCertificateValidationCallback(X509Certificate2Collection rootCertificates)
{
if (rootCertificates == null || rootCertificates.Count == 0)
{
return null;
}
return (sender, certificate, chain, sslPolicyErrors) =>
{
if ((sslPolicyErrors & SslPolicyErrors.RemoteCertificateNameMismatch) != 0 ||
(sslPolicyErrors & SslPolicyErrors.RemoteCertificateNotAvailable) != 0 ||
certificate == null)
{
return false;
}
var serverCertificate = certificate as X509Certificate2 ?? new X509Certificate2(certificate);
try
{
if (!HasServerAuthenticationEku(serverCertificate))
{
return false;
}
using var customChain = new X509Chain();
customChain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck;
customChain.ChainPolicy.VerificationFlags = X509VerificationFlags.AllowUnknownCertificateAuthority;
foreach (var rootCertificate in rootCertificates)
{
customChain.ChainPolicy.ExtraStore.Add(rootCertificate);
}
if (chain != null)
{
foreach (var chainElement in chain.ChainElements)
{
if (!string.Equals(chainElement.Certificate.Thumbprint, serverCertificate.Thumbprint, StringComparison.OrdinalIgnoreCase))
{
customChain.ChainPolicy.ExtraStore.Add(chainElement.Certificate);
}
}
}
return customChain.Build(serverCertificate) && ChainEndsWithTrustedRoot(customChain, rootCertificates);
}
finally
{
if (!ReferenceEquals(serverCertificate, certificate))
{
serverCertificate.Dispose();
}
}
};
}
private static bool HasServerAuthenticationEku(X509Certificate2 certificate)
{
var hasEnhancedKeyUsage = false;
foreach (var extension in certificate.Extensions)
{
if (extension is X509EnhancedKeyUsageExtension enhancedKeyUsage)
{
hasEnhancedKeyUsage = true;
foreach (var oid in enhancedKeyUsage.EnhancedKeyUsages)
{
if (string.Equals(oid.Value, ServerAuthenticationOid, StringComparison.Ordinal))
{
return true;
}
}
}
}
return !hasEnhancedKeyUsage;
}
private static bool ChainEndsWithTrustedRoot(X509Chain chain, X509Certificate2Collection rootCertificates)
{
if (chain.ChainElements.Count == 0)
{
return false;
}
var chainRoot = chain.ChainElements[chain.ChainElements.Count - 1].Certificate;
foreach (var rootCertificate in rootCertificates)
{
if (string.Equals(chainRoot.Thumbprint, rootCertificate.Thumbprint, StringComparison.OrdinalIgnoreCase))
{
return true;
}
}
return false;
}
private static LocalCertificateSelectionCallback CreateLocalCertificateSelectionCallback(X509Certificate2 clientCertificate)
{
if (clientCertificate == null)
{
return null;
}
return (sender, targetHost, localCertificates, remoteCertificate, acceptableIssuers) => clientCertificate;
}
public async Task<int> CreateDatabase(string dbName)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var status = await client.ServiceClient.setStorageGroupAsync(client.SessionId, dbName);
if (_debugMode)
{
_logger.LogInformation("create database {0} successfully, server message is {1}", dbName, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when creating database"
);
}
[Obsolete("This method is deprecated, please use createDatabase instead.")]
public async Task<int> SetStorageGroup(string groupName)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var status = await client.ServiceClient.setStorageGroupAsync(client.SessionId, groupName);
if (_debugMode)
{
_logger.LogInformation("set storage group {0} successfully, server message is {1}", groupName, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when setting storage group"
);
}
public async Task<int> CreateTimeSeries(
string tsPath,
TSDataType dataType,
TSEncoding encoding,
Compressor compressor)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = new TSCreateTimeseriesReq(
client.SessionId,
tsPath,
(int)dataType,
(int)encoding,
(int)compressor);
var status = await client.ServiceClient.createTimeseriesAsync(req);
if (_debugMode)
{
_logger.LogInformation("creating time series {0} successfully, server message is {1}", tsPath, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when creating time series"
);
}
public async Task<int> CreateAlignedTimeseriesAsync(
string prefixPath,
List<string> measurements,
List<TSDataType> dataTypeLst,
List<TSEncoding> encodingLst,
List<Compressor> compressorLst)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var dataTypes = dataTypeLst.ConvertAll(x => (int)x);
var encodings = encodingLst.ConvertAll(x => (int)x);
var compressors = compressorLst.ConvertAll(x => (int)x);
var req = new TSCreateAlignedTimeseriesReq(
client.SessionId,
prefixPath,
measurements,
dataTypes,
encodings,
compressors);
var status = await client.ServiceClient.createAlignedTimeseriesAsync(req);
if (_debugMode)
{
_logger.LogInformation("creating aligned time series {0} successfully, server message is {1}", prefixPath, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when creating aligned time series"
);
}
public async Task<int> DeleteDatabaseAsync(string dbName)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var status = await client.ServiceClient.deleteStorageGroupsAsync(client.SessionId, new List<string> { dbName });
if (_debugMode)
{
_logger.LogInformation("delete database {0} successfully, server message is {1}", dbName, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when deleting database"
);
}
[Obsolete("This method is deprecated, please use DeleteDatabaseAsync instead.")]
public async Task<int> DeleteStorageGroupAsync(string groupName)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var status = await client.ServiceClient.deleteStorageGroupsAsync(client.SessionId, new List<string> { groupName });
if (_debugMode)
{
_logger.LogInformation("delete storage group {0} successfully, server message is {1}", groupName, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when deleting storage group"
);
}
public async Task<int> DeleteDatabasesAsync(List<string> dbNames)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var status = await client.ServiceClient.deleteStorageGroupsAsync(client.SessionId, dbNames);
if (_debugMode)
{
_logger.LogInformation("delete database(s) {0} successfully, server message is {1}", dbNames, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when deleting database(s)"
);
}
[Obsolete("This method is deprecated, please use DeleteDatabasesAsync instead.")]
public async Task<int> DeleteStorageGroupsAsync(List<string> groupNames)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var status = await client.ServiceClient.deleteStorageGroupsAsync(client.SessionId, groupNames);
if (_debugMode)
{
_logger.LogInformation("delete storage group(s) {0} successfully, server message is {1}", groupNames, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when deleting storage group(s)"
);
}
public async Task<int> CreateMultiTimeSeriesAsync(
List<string> tsPathLst,
List<TSDataType> dataTypeLst,
List<TSEncoding> encodingLst,
List<Compressor> compressorLst)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var dataTypes = dataTypeLst.ConvertAll(x => (int)x);
var encodings = encodingLst.ConvertAll(x => (int)x);
var compressors = compressorLst.ConvertAll(x => (int)x);
var req = new TSCreateMultiTimeseriesReq(client.SessionId, tsPathLst, dataTypes, encodings, compressors);
var status = await client.ServiceClient.createMultiTimeseriesAsync(req);
if (_debugMode)
{
_logger.LogInformation("creating multiple time series {0}, server message is {1}", tsPathLst, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when creating multiple time series"
);
}
public async Task<int> DeleteTimeSeriesAsync(List<string> pathList)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var status = await client.ServiceClient.deleteTimeseriesAsync(client.SessionId, pathList);
if (_debugMode)
{
_logger.LogInformation("deleting multiple time series {0}, server message is {1}", pathList, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when deleting multiple time series"
);
}
public async Task<int> DeleteTimeSeriesAsync(string tsPath)
{
return await DeleteTimeSeriesAsync(new List<string> { tsPath });
}
public async Task<bool> CheckTimeSeriesExistsAsync(string tsPath)
{
try
{
var sql = "SHOW TIMESERIES " + tsPath;
var sessionDataSet = await ExecuteQueryStatementAsync(sql);
bool timeSeriesExists = await sessionDataSet.HasNextAsync();
await sessionDataSet.Close(); // be sure to close the SessionDataSet to put the client back to the pool
return timeSeriesExists;
}
catch (TException e)
{
throw new TException("could not check if certain time series exists", e);
}
}
public async Task<int> DeleteDataAsync(List<string> tsPathLst, long startTime, long endTime)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = new TSDeleteDataReq(client.SessionId, tsPathLst, startTime, endTime);
var status = await client.ServiceClient.deleteDataAsync(req);
if (_debugMode)
{
_logger.LogInformation(
"delete data from {0}, server message is {1}",
tsPathLst,
status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when deleting data"
);
}
public async Task<int> InsertRecordAsync(string deviceId, RowRecord record)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = new TSInsertRecordReq(client.SessionId, deviceId, record.Measurements, record.ToBytes(), record.Timestamps);
var status = await client.ServiceClient.insertRecordAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert one record to device {0}， server message: {1}", deviceId, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting record"
);
}
public async Task<int> InsertAlignedRecordAsync(string deviceId, RowRecord record)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = new TSInsertRecordReq(client.SessionId, deviceId, record.Measurements, record.ToBytes(), record.Timestamps);
req.IsAligned = true;
// ASSERT that the insert plan is aligned
System.Diagnostics.Debug.Assert(req.IsAligned == true);
var status = await client.ServiceClient.insertRecordAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert one record to device {0}， server message: {1}", deviceId, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting record"
);
}
public TSInsertStringRecordReq GenInsertStrRecordReq(string deviceId, List<string> measurements,
List<string> values, long timestamp, long sessionId, bool isAligned = false)
{
if (values.Count() != measurements.Count())
{
throw new ArgumentException("length of data types does not equal to length of values!");
}
return new TSInsertStringRecordReq(sessionId, deviceId, measurements, values, timestamp)
{
IsAligned = isAligned
};
}
public TSInsertStringRecordsReq GenInsertStringRecordsReq(List<string> deviceIds, List<List<string>> measurementsList,
List<List<string>> valuesList, List<long> timestamps, long sessionId, bool isAligned = false)
{
if (valuesList.Count() != measurementsList.Count())
{
throw new ArgumentException("length of data types does not equal to length of values!");
}
return new TSInsertStringRecordsReq(sessionId, deviceIds, measurementsList, valuesList, timestamps)
{
IsAligned = isAligned
};
}
public TSInsertRecordsReq GenInsertRecordsReq(List<string> deviceId, List<RowRecord> rowRecords,
long sessionId)
{
var measurementLst = rowRecords.Select(x => x.Measurements).ToList();
var timestampLst = rowRecords.Select(x => x.Timestamps).ToList();
var valuesLstInBytes = rowRecords.Select(row => row.ToBytes()).ToList();
return new TSInsertRecordsReq(sessionId, deviceId, measurementLst, valuesLstInBytes, timestampLst);
}
public async Task<int> InsertStringRecordAsync(string deviceId, List<string> measurements, List<string> values,
long timestamp)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertStrRecordReq(deviceId, measurements, values, timestamp, client.SessionId);
var status = await client.ServiceClient.insertStringRecordAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert one string record to device {0}， server message: {1}", deviceId, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting a string record"
);
}
public async Task<int> InsertAlignedStringRecordAsync(string deviceId, List<string> measurements, List<string> values,
long timestamp)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertStrRecordReq(deviceId, measurements, values, timestamp, client.SessionId, true);
var status = await client.ServiceClient.insertStringRecordAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert one record to device {0}， server message: {1}", deviceId, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting a string record"
);
}
public async Task<int> InsertStringRecordsAsync(List<string> deviceIds, List<List<string>> measurements, List<List<string>> values,
List<long> timestamps)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertStringRecordsReq(deviceIds, measurements, values, timestamps, client.SessionId);
var status = await client.ServiceClient.insertStringRecordsAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert string records to devices {0}， server message: {1}", deviceIds, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting string records"
);
}
public async Task<int> InsertAlignedStringRecordsAsync(List<string> deviceIds, List<List<string>> measurements, List<List<string>> values,
List<long> timestamps)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertStringRecordsReq(deviceIds, measurements, values, timestamps, client.SessionId, true);
var status = await client.ServiceClient.insertStringRecordsAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert string records to devices {0}， server message: {1}", deviceIds, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting string records"
);
}
public async Task<int> InsertRecordsAsync(List<string> deviceId, List<RowRecord> rowRecords)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertRecordsReq(deviceId, rowRecords, client.SessionId);
var status = await client.ServiceClient.insertRecordsAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert multiple records to devices {0}, server message: {1}", deviceId, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting records"
);
}
public async Task<int> InsertAlignedRecordsAsync(List<string> deviceId, List<RowRecord> rowRecords)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertRecordsReq(deviceId, rowRecords, client.SessionId);
req.IsAligned = true;
// ASSERT that the insert plan is aligned
System.Diagnostics.Debug.Assert(req.IsAligned == true);
var status = await client.ServiceClient.insertRecordsAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert multiple records to devices {0}, server message: {1}", deviceId, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting records"
);
}
public TSInsertTabletReq GenInsertTabletReq(Tablet tablet, long sessionId)
{
return new TSInsertTabletReq(
sessionId,
tablet.InsertTargetName,
tablet.Measurements,
tablet.GetBinaryValues(),
tablet.GetBinaryTimestamps(),
tablet.GetDataTypes(),
tablet.RowNumber);
}
public async Task<int> InsertTabletAsync(Tablet tablet)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertTabletReq(tablet, client.SessionId);
var status = await client.ServiceClient.insertTabletAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert one tablet to device {0}, server message: {1}", tablet.InsertTargetName, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting tablet"
);
}
public async Task<int> InsertAlignedTabletAsync(Tablet tablet)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertTabletReq(tablet, client.SessionId);
req.IsAligned = true;
var status = await client.ServiceClient.insertTabletAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert one aligned tablet to device {0}, server message: {1}", tablet.InsertTargetName, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting aligned tablet"
);
}
protected internal async Task<int> InsertRelationalTabletAsync(Tablet tablet)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertTabletReq(tablet, client.SessionId);
req.ColumnCategories = tablet.GetColumnColumnCategories();
req.WriteToTable = true;
var status = await client.ServiceClient.insertTabletAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert one tablet to table {0}, server message: {1}", tablet.InsertTargetName, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting tablet"
);
}
public TSInsertTabletsReq GenInsertTabletsReq(List<Tablet> tabletLst, long sessionId)
{
var deviceIdLst = new List<string>();
var measurementsLst = new List<List<string>>();
var valuesLst = new List<byte[]>();
var timestampsLst = new List<byte[]>();
var typeLst = new List<List<int>>();
var sizeLst = new List<int>();
foreach (var tablet in tabletLst)
{
var dataTypeValues = tablet.GetDataTypes();
deviceIdLst.Add(tablet.InsertTargetName);
measurementsLst.Add(tablet.Measurements);
valuesLst.Add(tablet.GetBinaryValues());
timestampsLst.Add(tablet.GetBinaryTimestamps());
typeLst.Add(dataTypeValues);
sizeLst.Add(tablet.RowNumber);
}
return new TSInsertTabletsReq(
sessionId,
deviceIdLst,
measurementsLst,
valuesLst,
timestampsLst,
typeLst,
sizeLst);
}
public async Task<int> InsertTabletsAsync(List<Tablet> tabletLst)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertTabletsReq(tabletLst, client.SessionId);
var status = await client.ServiceClient.insertTabletsAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert multiple tablets, message: {0}", status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting tablets"
);
}
public async Task<int> InsertAlignedTabletsAsync(List<Tablet> tabletLst)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertTabletsReq(tabletLst, client.SessionId);
req.IsAligned = true;
var status = await client.ServiceClient.insertTabletsAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert multiple aligned tablets, message: {0}", status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting aligned tablets"
);
}
public async Task<int> InsertRecordsOfOneDeviceAsync(string deviceId, List<RowRecord> rowRecords)
{
var sortedRowRecords = rowRecords.OrderBy(x => x.Timestamps).ToList();
return await InsertRecordsOfOneDeviceSortedAsync(deviceId, sortedRowRecords);
}
public async Task<int> InsertAlignedRecordsOfOneDeviceAsync(string deviceId, List<RowRecord> rowRecords)
{
var sortedRowRecords = rowRecords.OrderBy(x => x.Timestamps).ToList();
return await InsertAlignedRecordsOfOneDeviceSortedAsync(deviceId, sortedRowRecords);
}
public async Task<int> InsertStringRecordsOfOneDeviceAsync(string deviceId, List<long> timestamps,
List<List<string>> measurementsList, List<List<string>> valuesList)
{
var joined = timestamps.Zip(measurementsList, (t, m) => new { t, m })
.Zip(valuesList, (tm, v) => new { tm.t, tm.m, v })
.OrderBy(x => x.t);
var sortedTimestamps = joined.Select(x => x.t).ToList();
var sortedMeasurementsList = joined.Select(x => x.m).ToList();
var sortedValuesList = joined.Select(x => x.v).ToList();
return await InsertStringRecordsOfOneDeviceSortedAsync(deviceId, sortedTimestamps, sortedMeasurementsList, sortedValuesList, false);
}
public async Task<int> InsertAlignedStringRecordsOfOneDeviceAsync(string deviceId, List<long> timestamps,
List<List<string>> measurementsList, List<List<string>> valuesList)
{
var joined = timestamps.Zip(measurementsList, (t, m) => new { t, m })
.Zip(valuesList, (tm, v) => new { tm.t, tm.m, v })
.OrderBy(x => x.t);
var sortedTimestamps = joined.Select(x => x.t).ToList();
var sortedMeasurementsList = joined.Select(x => x.m).ToList();
var sortedValuesList = joined.Select(x => x.v).ToList();
return await InsertStringRecordsOfOneDeviceSortedAsync(deviceId, sortedTimestamps, sortedMeasurementsList, sortedValuesList, true);
}
public async Task<int> InsertStringRecordsOfOneDeviceSortedAsync(string deviceId, List<long> timestamps,
List<List<string>> measurementsList, List<List<string>> valuesList, bool isAligned)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
if (!_utilFunctions.IsSorted(timestamps))
{
throw new ArgumentException("insert string records of one device error: timestamp not sorted");
}
var req = GenInsertStringRecordsOfOneDeviceReq(deviceId, timestamps, measurementsList, valuesList, client.SessionId, isAligned);
var status = await client.ServiceClient.insertStringRecordsOfOneDeviceAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert string records of one device, message: {0}", status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting string records of one device"
);
}
private TSInsertStringRecordsOfOneDeviceReq GenInsertStringRecordsOfOneDeviceReq(string deviceId,
List<long> timestamps, List<List<string>> measurementsList, List<List<string>> valuesList,
long sessionId, bool isAligned = false)
{
return new TSInsertStringRecordsOfOneDeviceReq(
sessionId,
deviceId,
measurementsList,
valuesList,
timestamps)
{
IsAligned = isAligned
};
}
private TSInsertRecordsOfOneDeviceReq GenInsertRecordsOfOneDeviceRequest(
string deviceId,
List<RowRecord> records,
long sessionId)
{
var values = records.Select(row => row.ToBytes());
var measurementsLst = records.Select(x => x.Measurements).ToList();
var timestampLst = records.Select(x => x.Timestamps).ToList();
return new TSInsertRecordsOfOneDeviceReq(
sessionId,
deviceId,
measurementsLst,
values.ToList(),
timestampLst);
}
public async Task<int> InsertRecordsOfOneDeviceSortedAsync(string deviceId, List<RowRecord> rowRecords)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var timestampLst = rowRecords.Select(x => x.Timestamps).ToList();
if (!_utilFunctions.IsSorted(timestampLst))
{
throw new ArgumentException("insert records of one device error: timestamp not sorted");
}
var req = GenInsertRecordsOfOneDeviceRequest(deviceId, rowRecords, client.SessionId);
var status = await client.ServiceClient.insertRecordsOfOneDeviceAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert records of one device, message: {0}", status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting records of one device"
);
}
public async Task<int> InsertAlignedRecordsOfOneDeviceSortedAsync(string deviceId, List<RowRecord> rowRecords)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var timestampLst = rowRecords.Select(x => x.Timestamps).ToList();
if (!_utilFunctions.IsSorted(timestampLst))
{
throw new ArgumentException("insert records of one device error: timestamp not sorted");
}
var req = GenInsertRecordsOfOneDeviceRequest(deviceId, rowRecords, client.SessionId);
req.IsAligned = true;
var status = await client.ServiceClient.insertRecordsOfOneDeviceAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert aligned records of one device, message: {0}", status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when inserting aligned records of one device"
);
}
public async Task<int> TestInsertRecordAsync(string deviceId, RowRecord record)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = new TSInsertRecordReq(
client.SessionId,
deviceId,
record.Measurements,
record.ToBytes(),
record.Timestamps);
var status = await client.ServiceClient.testInsertRecordAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert one record to device {0}， server message: {1}", deviceId, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when test inserting one record"
);
}
public async Task<int> TestInsertRecordsAsync(List<string> deviceId, List<RowRecord> rowRecords)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertRecordsReq(deviceId, rowRecords, client.SessionId);
var status = await client.ServiceClient.testInsertRecordsAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert multiple records to devices {0}, server message: {1}", deviceId, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when test inserting multiple records"
);
}
public async Task<int> TestInsertTabletAsync(Tablet tablet)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertTabletReq(tablet, client.SessionId);
var status = await client.ServiceClient.testInsertTabletAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert one tablet to device {0}, server message: {1}", tablet.InsertTargetName, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when test inserting one tablet"
);
}
public async Task<int> TestInsertTabletsAsync(List<Tablet> tabletLst)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = GenInsertTabletsReq(tabletLst, client.SessionId);
var status = await client.ServiceClient.testInsertTabletsAsync(req);
if (_debugMode)
{
_logger.LogInformation("insert multiple tablets, message: {0}", status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when test inserting multiple tablets"
);
}
public async Task<SessionDataSet> ExecuteQueryStatementAsync(string sql)
{
// default timeout is 60s
return await ExecuteQueryStatementAsync(sql, 60 * 1000);
}
public async Task<SessionDataSet> ExecuteQueryStatementAsync(string sql, long timeoutInMs)
{
return await ExecuteClientOperationAsync<SessionDataSet>(
async client =>
{
var req = new TSExecuteStatementReq(client.SessionId, sql, client.StatementId)
{
FetchSize = _fetchSize,
Timeout = timeoutInMs
};
var resp = await client.ServiceClient.executeQueryStatementV2Async(req);
var status = resp.Status;
if (_utilFunctions.VerifySuccess(status) == -1)
{
throw new Exception(string.Format("execute query failed, sql: {0}, message: {1}", sql, status.Message));
}
return new SessionDataSet(
sql, resp.Columns, resp.DataTypeList, resp.ColumnNameIndexMap, resp.QueryId,
client.StatementId, client, resp.QueryResult, resp.IgnoreTimeStamp,
resp.MoreData, _zoneId, resp.ColumnIndex2TsBlockColumnIndexList, _clients
)
{
FetchSize = _fetchSize,
};
},
errMsg: "Error occurs when executing query statement",
putClientBack: false
);
}
public async Task<SessionDataSet> ExecuteStatementAsync(string sql, long timeout)
{
return await ExecuteClientOperationAsync<SessionDataSet>(
async client =>
{
var req = new TSExecuteStatementReq(client.SessionId, sql, client.StatementId)
{
FetchSize = _fetchSize,
Timeout = timeout
};
var resp = await client.ServiceClient.executeStatementV2Async(req);
var status = resp.Status;
if (_utilFunctions.VerifySuccess(status) == -1)
{
throw new Exception(string.Format("execute query failed, sql: {0}, message: {1}", sql, status.Message));
}
return new SessionDataSet(
sql, resp.Columns, resp.DataTypeList, resp.ColumnNameIndexMap, resp.QueryId,
client.StatementId, client, resp.QueryResult, resp.IgnoreTimeStamp,
resp.MoreData, _zoneId, resp.ColumnIndex2TsBlockColumnIndexList, _clients
)
{
FetchSize = _fetchSize,
};
},
errMsg: "Error occurs when executing query statement",
putClientBack: false
);
}
public async Task<int> ExecuteNonQueryStatementAsync(string sql)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
String previousDB = _database;
var req = new TSExecuteStatementReq(client.SessionId, sql, client.StatementId);
var resp = await client.ServiceClient.executeUpdateStatementAsync(req);
var status = resp.Status;
if (resp.__isset.database)
{
this._database = resp.Database;
}
if (_database != previousDB)
{
// all client should switch to the same database
var failedClients = new List<(long SessionId, Exception Error)>();
foreach (var c in _clients.ClientQueue.AsEnumerable())
{
try
{
if (c != client)
{
var switchReq = new TSExecuteStatementReq(c.SessionId, sql, c.StatementId);
await c.ServiceClient.executeUpdateStatementAsync(switchReq);
}
}
catch (Exception e)
{
failedClients.Add((c.SessionId, e));
_logger?.LogError("switch database from {0} to {1} failed for {2}, error: {3}", previousDB, _database, c.SessionId, e.Message);
}
}
if (failedClients.Count > 0)
{
throw new TException($"Database switch partially failed: {failedClients.Count} client(s) could not switch from {previousDB} to {_database}", failedClients[0].Error);
}
_logger?.LogInformation("switch database from {0} to {1}", previousDB, _database);
}
if (_debugMode)
{
_logger.LogInformation("execute non-query statement {0} message: {1}", sql, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when executing non-query statement"
);
}
public async Task<SessionDataSet> ExecuteRawDataQuery(List<string> paths, long startTime, long endTime)
{
return await ExecuteClientOperationAsync<SessionDataSet>(
async client =>
{
var req = new TSRawDataQueryReq(client.SessionId, paths, startTime, endTime, client.StatementId)
{
FetchSize = _fetchSize,
EnableRedirectQuery = false
};
var resp = await client.ServiceClient.executeRawDataQueryV2Async(req);
var status = resp.Status;
if (_utilFunctions.VerifySuccess(status) == -1)
{
throw new Exception(string.Format("execute raw data query failed, message: {0}", status.Message));
}
return new SessionDataSet(
"", resp.Columns, resp.DataTypeList, resp.ColumnNameIndexMap, resp.QueryId,
client.StatementId, client, resp.QueryResult, resp.IgnoreTimeStamp,
resp.MoreData, _zoneId, resp.ColumnIndex2TsBlockColumnIndexList, _clients
)
{
FetchSize = _fetchSize,
};
},
errMsg: "Error occurs when executing raw data query",
putClientBack: false
);
}
public async Task<SessionDataSet> ExecuteLastDataQueryAsync(List<string> paths, long lastTime)
{
return await ExecuteClientOperationAsync<SessionDataSet>(
async client =>
{
var req = new TSLastDataQueryReq(client.SessionId, paths, lastTime, client.StatementId)
{
FetchSize = _fetchSize,
EnableRedirectQuery = false
};
var resp = await client.ServiceClient.executeLastDataQueryV2Async(req);
var status = resp.Status;
if (_utilFunctions.VerifySuccess(status) == -1)
{
throw new Exception(string.Format("execute last data query failed, message: {0}", status.Message));
}
return new SessionDataSet(
"", resp.Columns, resp.DataTypeList, resp.ColumnNameIndexMap, resp.QueryId,
client.StatementId, client, resp.QueryResult, resp.IgnoreTimeStamp,
resp.MoreData, _zoneId, resp.ColumnIndex2TsBlockColumnIndexList, _clients
)
{
FetchSize = _fetchSize,
};
},
errMsg: "Error occurs when executing last data query",
putClientBack: false
);
}
[Obsolete("This method is obsolete. Use SQL instead.", false)]
public async Task<int> CreateSchemaTemplateAsync(Template template)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = new TSCreateSchemaTemplateReq(client.SessionId, template.Name, template.ToBytes());
var status = await client.ServiceClient.createSchemaTemplateAsync(req);
if (_debugMode)
{
_logger.LogInformation("create schema template {0} message: {1}", template.Name, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when creating schema template"
);
}
[Obsolete("This method is obsolete. Use SQL instead.", false)]
public async Task<int> DropSchemaTemplateAsync(string templateName)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = new TSDropSchemaTemplateReq(client.SessionId, templateName);
var status = await client.ServiceClient.dropSchemaTemplateAsync(req);
if (_debugMode)
{
_logger.LogInformation("drop schema template {0} message: {1}", templateName, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when dropping schema template"
);
}
[Obsolete("This method is obsolete. Use SQL instead.", false)]
public async Task<int> SetSchemaTemplateAsync(string templateName, string prefixPath)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = new TSSetSchemaTemplateReq(client.SessionId, templateName, prefixPath);
var status = await client.ServiceClient.setSchemaTemplateAsync(req);
if (_debugMode)
{
_logger.LogInformation("set schema template {0} message: {1}", templateName, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when setting schema template"
);
}
[Obsolete("This method is obsolete. Use SQL instead.", false)]
public async Task<int> UnsetSchemaTemplateAsync(string prefixPath, string templateName)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = new TSUnsetSchemaTemplateReq(client.SessionId, prefixPath, templateName);
var status = await client.ServiceClient.unsetSchemaTemplateAsync(req);
if (_debugMode)
{
_logger.LogInformation("unset schema template {0} message: {1}", templateName, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when unsetting schema template"
);
}
[Obsolete("This method is obsolete. Use SQL instead.", false)]
public async Task<int> DeleteNodeInTemplateAsync(string templateName, string path)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = new TSPruneSchemaTemplateReq(client.SessionId, templateName, path);
var status = await client.ServiceClient.pruneSchemaTemplateAsync(req);
if (_debugMode)
{
_logger.LogInformation("delete node in template {0} message: {1}", templateName, status.Message);
}
return _utilFunctions.VerifySuccess(status);
},
errMsg: "Error occurs when deleting node in template"
);
}
[Obsolete("This method is obsolete. Use SQL instead.", false)]
public async Task<int> CountMeasurementsInTemplateAsync(string name)
{
return await ExecuteClientOperationAsync<int>(
async client =>
{
var req = new TSQueryTemplateReq(client.SessionId, name, (int)TemplateQueryType.COUNT_MEASUREMENTS);
var resp = await client.ServiceClient.querySchemaTemplateAsync(req);
var status = resp.Status;
if (_debugMode)
{
_logger.LogInformation("count measurements in template {0} message: {1}", name, status.Message);
}
if (_utilFunctions.VerifySuccess(status) == -1)
{
throw new Exception(string.Format("count measurements in template failed, template name: {0}, message: {1}", name, status.Message));
}
return resp.Count;
},
errMsg: "Error occurs when counting measurements in template"
);
}
[Obsolete("This method is obsolete. Use SQL instead.", false)]
public async Task<bool> IsMeasurementInTemplateAsync(string templateName, string path)
{
return await ExecuteClientOperationAsync<bool>(
async client =>
{
var req = new TSQueryTemplateReq(client.SessionId, templateName, (int)TemplateQueryType.IS_MEASUREMENT);
req.Measurement = path;
var resp = await client.ServiceClient.querySchemaTemplateAsync(req);
var status = resp.Status;
if (_debugMode)
{
_logger.LogInformation("is measurement in template {0} message: {1}", templateName, status.Message);
}
if (_utilFunctions.VerifySuccess(status) == -1)
{
throw new Exception(string.Format("is measurement in template failed, template name: {0}, message: {1}", templateName, status.Message));
}
return resp.Result;
},
errMsg: "Error occurs when checking measurement in template"
);
}
[Obsolete("This method is obsolete. Use SQL instead.", false)]
public async Task<bool> IsPathExistInTemplateAsync(string templateName, string path)
{
return await ExecuteClientOperationAsync<bool>(
async client =>
{
var req = new TSQueryTemplateReq(client.SessionId, templateName, (int)TemplateQueryType.PATH_EXIST);
req.Measurement = path;
var resp = await client.ServiceClient.querySchemaTemplateAsync(req);
var status = resp.Status;
if (_debugMode)
{
_logger.LogInformation("is path exist in template {0} message: {1}", templateName, status.Message);
}
if (_utilFunctions.VerifySuccess(status) == -1)
{
throw new Exception(string.Format("is path exist in template failed, template name: {0}, message: {1}", templateName, status.Message));
}
return resp.Result;
},
errMsg: "Error occurs when checking path exist in template"
);
}
[Obsolete("This method is obsolete. Use SQL instead.", false)]
public async Task<List<string>> ShowMeasurementsInTemplateAsync(string templateName, string pattern = "")
{
return await ExecuteClientOperationAsync<List<string>>(
async client =>
{
var req = new TSQueryTemplateReq(client.SessionId, templateName, (int)TemplateQueryType.SHOW_MEASUREMENTS);
req.Measurement = pattern;
var resp = await client.ServiceClient.querySchemaTemplateAsync(req);
var status = resp.Status;
if (_debugMode)
{
_logger.LogInformation("get measurements in template {0} message: {1}", templateName, status.Message);
}
if (_utilFunctions.VerifySuccess(status) == -1)
{
throw new Exception(string.Format("show measurements in template failed, template name: {0}, pattern: {1}, message: {2}", templateName, pattern, status.Message));
}
return resp.Measurements;
},
errMsg: "Error occurs when showing measurements in template"
);
}
[Obsolete("This method is obsolete. Use SQL instead.", false)]
public async Task<List<string>> ShowAllTemplatesAsync()
{
return await ExecuteClientOperationAsync<List<string>>(
async client =>
{
var req = new TSQueryTemplateReq(client.SessionId, "", (int)TemplateQueryType.SHOW_TEMPLATES);
var resp = await client.ServiceClient.querySchemaTemplateAsync(req);
var status = resp.Status;
if (_debugMode)
{
_logger.LogInformation("get all templates message: {0}", status.Message);
}
if (_utilFunctions.VerifySuccess(status) == -1)
{
throw new Exception(string.Format("show all templates failed, message: {0}", status.Message));
}
return resp.Measurements;
},
errMsg: "Error occurs when getting all templates"
);
}
[Obsolete("This method is obsolete. Use SQL instead.", false)]
public async Task<List<string>> ShowPathsTemplateSetOnAsync(string templateName)
{
return await ExecuteClientOperationAsync<List<string>>(
async client =>
{
var req = new TSQueryTemplateReq(client.SessionId, templateName, (int)TemplateQueryType.SHOW_SET_TEMPLATES);
var resp = await client.ServiceClient.querySchemaTemplateAsync(req);
var status = resp.Status;
if (_debugMode)
{
_logger.LogInformation("get paths template set on {0} message: {1}", templateName, status.Message);
}
if (_utilFunctions.VerifySuccess(status) == -1)
{
throw new Exception(string.Format("show paths template set on failed, template name: {0}, message: {1}", templateName, status.Message));
}
return resp.Measurements;
},
errMsg: "Error occurs when getting paths template set on"
);
}
[Obsolete("This method is obsolete. Use SQL instead.", false)]
public async Task<List<string>> ShowPathsTemplateUsingOnAsync(string templateName)
{
return await ExecuteClientOperationAsync<List<string>>(
async client =>
{
var req = new TSQueryTemplateReq(client.SessionId, templateName, (int)TemplateQueryType.SHOW_USING_TEMPLATES);
var resp = await client.ServiceClient.querySchemaTemplateAsync(req);
var status = resp.Status;
if (_debugMode)
{
_logger.LogInformation("get paths template using on {0} message: {1}", templateName, status.Message);
}
if (_utilFunctions.VerifySuccess(status) == -1)
{
throw new Exception(string.Format("show paths template using on failed, template name: {0}, message: {1}", templateName, status.Message));
}
return resp.Measurements;
},
errMsg: "Error occurs when getting paths template using on"
);
}
protected virtual void Dispose(bool disposing)
{
if (!disposedValue)
{
if (disposing)
{
#if NET461_OR_GREATER || NETSTANDARD2_0
#else
_clients?.ClientQueue.Clear();
#endif
_clientCertificate?.Dispose();
if (_rootCertificates != null)
{
foreach (var rootCertificate in _rootCertificates)
{
rootCertificate.Dispose();
}
}
}
_clients = null;
disposedValue = true;
}
}
public void Dispose()
{
Dispose(disposing: true);
GC.SuppressFinalize(this);
}
SessionPoolDepletedException IPoolDiagnosticReporter.BuildDepletionException(string reasonPhrase)
=> new SessionPoolDepletedException(reasonPhrase, AvailableClients, TotalPoolSize, FailedReconnections);
}
}