Kinetica.cs

Go to the documentation of this file.

using Avro.IO;
using Newtonsoft.Json;
using Snappier;
using System.Net;
using System.Net.Http.Headers;
using System.Text;
using kinetica.Utils;


namespace kinetica;

public partial class Kinetica : IDisposable
    {
        private bool _disposed = false;
        public const int END_OF_SET = -9999;

        public class Options
        {
            public string Username { get; set; } = string.Empty;

            public string Password { get; set; } = string.Empty;

            public string OauthToken { get; set; } = string.Empty;

            public bool UseSnappy { get; set; } = false;

            public int ThreadCount { get; set; } = 1;

            public bool DisableFailover { get; set; } = false;

            public bool DisableAutoDiscovery { get; set; } = false;

            public HAFailoverOrder HAFailoverOrder { get; set; } = HAFailoverOrder.Random;

            public int Timeout { get; set; } = 0;

            public int HostManagerPort { get; set; } = HAFailoverManager.DefaultHostManagerPort;

            public string? HostnameRegex { get; set; } = null;

            public string PrimaryUrl { get; set; } = string.Empty;

            public int InitialConnectionAttemptTimeout { get; set; } = 0;

            public int ServerConnectionTimeout { get; set; } = 60000;

            public TimeSpan PooledConnectionLifetime { get; set; } = TimeSpan.FromMinutes(2);

            public TimeSpan PooledConnectionIdleTimeout { get; set; } = TimeSpan.FromMinutes(2);
        }

        public static string GetApiVersion() { return API_VERSION; }

        public string Url { get; private set; }

        public Uri URL { get; private set; }

        public string? Username { get; private set; } = null;

        private string? Password { get; set; } = null;

        private string? OauthToken { get; set; } = null;

        private string? Authorization { get; set; } = null;

        public bool UseSnappy { get; set; } = false;

        public int ThreadCount { get; set; } = 1;

        private HAFailoverManager? _haFailoverManager = null;

        public HAFailoverManager? HAManager => _haFailoverManager;

        public int NumClusterSwitches => _haFailoverManager?.NumClusterSwitches ?? 0;

        public IList<ClusterAddressInfo> GetHARingInfo()
        {
            return _haFailoverManager?.GetHostAddresses() ?? new List<ClusterAddressInfo>();
        }

        public ClusterAddressInfo? GetCurrentClusterInfo()
        {
            return _haFailoverManager?.GetClusterInfo();
        }

        // private string authorization;
        private volatile System.Collections.Concurrent.ConcurrentDictionary<string, KineticaType> knownTypes = new();

        // private type label to type ID lookup table
        private Dictionary<string, string> typeNameLookup = [];

        private readonly IHttpTransport _transport;

        // private object class type to KineticaType lookup table
        private Dictionary<Type, KineticaType> kineticaTypeLookup = [];

        internal Kinetica(string url_str, IHttpTransport transport, Options? options = null)
            : this(new List<string> { url_str }, transport, options)
        {
        }

        internal Kinetica(IList<string> urls, IHttpTransport transport, Options? options = null)
        {
            if (urls == null || urls.Count == 0)
                throw new KineticaException("At least one URL must be provided");

            // Use the first URL as the primary
            Url = urls[0].TrimEnd('/');
            URL = new Uri(Url);

            // Use the provided transport (for testing)
            _transport = transport;

            // Initialize other properties from options
            if (options != null)
            {
                Username = options.Username;
                Password = options.Password;
                OauthToken = options.OauthToken;
                UseSnappy = options.UseSnappy;
                ThreadCount = options.ThreadCount;

                // Create authorization header
                Authorization = CreateAuthorizationHeader();

                // Initialize HA failover manager if multiple URLs or options require it
                if (urls.Count > 1 || !options.DisableFailover)
                {
                    _haFailoverManager = new HAFailoverManager
                    {
                        DisableFailover = options.DisableFailover,
                        DisableAutoDiscovery = options.DisableAutoDiscovery,
                        HostManagerPort = options.HostManagerPort,
                        FailoverOrder = options.HAFailoverOrder
                    };

                    if (!string.IsNullOrEmpty(options.HostnameRegex))
                    {
                        _haFailoverManager.HostnameRegex = new System.Text.RegularExpressions.Regex(options.HostnameRegex);
                    }

                    var uriList = urls.Select(u => new Uri(u.TrimEnd('/'))).ToList();
                    _haFailoverManager.Initialize(uriList, this);
                }
            }
            else
            {
                _haFailoverManager = new HAFailoverManager { DisableAutoDiscovery = true };
                var uriList = urls.Select(u => new Uri(u.TrimEnd('/'))).ToList();
                _haFailoverManager.Initialize(uriList, null);
            }
        }

        public Kinetica( string url_str, Options? options = null )
            : this(new List<string> { url_str }, options)
        {
        }

        public Kinetica( IList<string> urls, Options? options = null )
        {
            if (urls == null || urls.Count == 0)
                throw new KineticaException("At least one URL must be provided");

            // Use the first URL as the primary
            Url = urls[0].TrimEnd('/');
            URL = new Uri(Url);

            // Initialize HTTP transport layer
            var timeout = options?.Timeout > 0
                ? TimeSpan.FromMilliseconds(options.Timeout)
                : TimeSpan.FromSeconds(30); // Default timeout

            _transport = new HttpClientTransport(
                timeout,
                options?.PooledConnectionLifetime,
                options?.PooledConnectionIdleTimeout);

            if ( null != options ) // If caller specified options
            {
                Username = options.Username;
                Password = options.Password;
                OauthToken = options.OauthToken;

                // Handle authorization
                Authorization = CreateAuthorizationHeader();

                UseSnappy = options.UseSnappy;
                ThreadCount = options.ThreadCount;

                // Initialize HA failover manager if there are multiple URLs or HA options are set
                if (urls.Count > 1 || !options.DisableAutoDiscovery)
                {
                    _haFailoverManager = new HAFailoverManager
                    {
                        DisableFailover = options.DisableFailover,
                        DisableAutoDiscovery = options.DisableAutoDiscovery,
                        HostManagerPort = options.HostManagerPort,
                        FailoverOrder = options.HAFailoverOrder
                    };

                    if (!string.IsNullOrEmpty(options.HostnameRegex))
                    {
                        _haFailoverManager.HostnameRegex = new System.Text.RegularExpressions.Regex(options.HostnameRegex);
                    }

                    // Convert string URLs to Uri objects
                    var uriList = urls.Select(u => new Uri(u.TrimEnd('/'))).ToList();

                    // Initialize with retry logic and exponential backoff (matching Rust implementation)
                    InitializeWithRetry(uriList, options);

                    // Update the URL to the current active cluster
                    var currentUrl = _haFailoverManager.GetUrl();
                    if (currentUrl != null)
                    {
                        Url = currentUrl.ToString().TrimEnd('/');
                        URL = currentUrl;
                    }
                }
            }
            else
            {
                // No options provided, initialize with a single URL
                _haFailoverManager = new HAFailoverManager
                {
                    DisableAutoDiscovery = true
                };
                var uriList = urls.Select(u => new Uri(u.TrimEnd('/'))).ToList();
                _haFailoverManager.Initialize(uriList, null);
            }
        }

        private void InitializeWithRetry(IList<Uri> uriList, Options options)
        {
            if (_haFailoverManager == null)
                throw new InvalidOperationException("HAFailoverManager not initialized");

            var startTime = DateTime.UtcNow;
            int attemptNumber = 0;
            int baseTimeoutMs = options.ServerConnectionTimeout > 0 ? options.ServerConnectionTimeout : 60000;
            int maxTotalTimeMs = options.InitialConnectionAttemptTimeout;
            Exception? lastException = null;

            while (true)
            {
                attemptNumber++;
                int currentTimeout = baseTimeoutMs * (1 << Math.Min(attemptNumber - 1, 5)); // Exponential backoff, cap at 32x

                try
                {
                    // Try to initialize with auto-discovery
                    _haFailoverManager.Initialize(uriList, this);
                    return; // Success
                }
                catch (Exception ex)
                {
                    lastException = ex;

                    // Check if this is a non-retriable error (like hostname regex mismatch or authorization failure)
                    if (IsNonRetriableInitializationError(ex))
                    {
                        // Fall back to initialization without auto-discovery
                        try
                        {
                            _haFailoverManager.DisableAutoDiscovery = true;
                            _haFailoverManager.Initialize(uriList, null);
                            return;
                        }
                        catch
                        {
                            throw new KineticaException($"Failed to initialize connection: {ex.Message}", ex);
                        }
                    }

                    // Check if we've exceeded the total timeout
                    var elapsed = (DateTime.UtcNow - startTime).TotalMilliseconds;
                    if (maxTotalTimeMs <= 0 || elapsed >= maxTotalTimeMs)
                    {
                        // No more retries - fall back to initialization without auto-discovery
                        try
                        {
                            _haFailoverManager.DisableAutoDiscovery = true;
                            _haFailoverManager.Initialize(uriList, null);
                            return;
                        }
                        catch
                        {
                            throw new KineticaException($"Failed to initialize connection after {attemptNumber} attempts: {ex.Message}", ex);
                        }
                    }

                    // Wait before retrying (exponential backoff)
                    int waitTime = Math.Min(currentTimeout, (int)(maxTotalTimeMs - elapsed));
                    if (waitTime > 0)
                    {
                        Thread.Sleep(Math.Min(waitTime, 5000)); // Cap wait at 5 seconds per attempt
                    }
                }
            }
        }

        private static bool IsNonRetriableInitializationError(Exception ex)
        {
            // Hostname regex failures won't change with retries
            if (ex.Message.Contains("hostname", StringComparison.OrdinalIgnoreCase) &&
                ex.Message.Contains("regex", StringComparison.OrdinalIgnoreCase))
            {
                return true;
            }

            // Authorization failures won't change with retries
            if (ex.Message.Contains("Unauthorized", StringComparison.OrdinalIgnoreCase) ||
                ex.Message.Contains("401", StringComparison.OrdinalIgnoreCase) ||
                ex.Message.Contains("credentials", StringComparison.OrdinalIgnoreCase))
            {
                return true;
            }

            return false;
        }

        public IList<Uri>? GetCurrentWorkerUrls()
        {
            var clusterInfo = _haFailoverManager?.GetClusterInfo();
            if (clusterInfo?.WorkerRankUrls != null && clusterInfo.WorkerRankUrls.Count > 0)
            {
                return clusterInfo.WorkerRankUrls;
            }

            // Try to get fresh worker URLs from the server
            try
            {
                var workers = new WorkerList(this);
                if (workers.Count > 0)
                {
                    return workers.ToList();
                }
            }
            catch
            {
                // Fall back to cached URLs or null
            }

            return null;
        }

        public IList<int>? GetCurrentRoutingTable()
        {
            try
            {
                return adminShowShards().rank;
            }
            catch
            {
                return null;
            }
        }

        public bool RefreshClusterInfo()
        {
            if (_haFailoverManager == null)
                return false;

            try
            {
                // Get fresh system properties to update cluster info
                var clusterInfo = _haFailoverManager.GetClusterInfo();
                if (clusterInfo == null)
                    return false;

                // Update system properties from the server
                var systemProps = showSystemProperties().property_map;
                clusterInfo.SystemProperties = systemProps;

                // Update worker URLs if multi-head is enabled
                var workers = new WorkerList(this);
                if (workers.Count > 0)
                {
                    clusterInfo.WorkerRankUrls = workers.ToList();
                }

                return true;
            }
            catch
            {
                return false;
            }
        }

        internal string? CreateAuthorizationHeader() {
            string? authorization = null;
            // Handle authorization
            if( OauthToken != null && OauthToken.Length > 0 ) {
                authorization = "Bearer " + OauthToken;
            }
            else if ( ( Username != null && ( Username.Length > 0 ) ) || ( Password != null && ( Password.Length > 0 ) ) )
            {
                authorization = ( "Basic " +
                                    Convert.ToBase64String( Encoding.GetEncoding( "ISO-8859-1" ).GetBytes( Username + ":" + Password ) ) );
            }

            return authorization;
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (!_disposed)
            {
                if (disposing)
                {
                    // Dispose managed resources
                    if (_transport is IDisposable disposable)
                    {
                        disposable.Dispose();
                    }
                }

                _disposed = true;
            }
        }

        public void AddTableType( string table_name, Type obj_type )
        {
            try
            {
                // Get the type from the table
                KineticaType ktype = KineticaType.fromTable( this, table_name );
                if ( ktype.getTypeID() == null )
                    throw new KineticaException( $"Could not get type ID for table '{table_name}'" );
                this.knownTypes.TryAdd( ktype.getTypeID(), ktype );

                // Save a mapping of the object to the KineticaType
                if ( obj_type != null )
                    this.SetKineticaSourceClassToTypeMapping( obj_type, ktype );

            } catch ( KineticaException ex )
            {
                throw new KineticaException( "Error creating type from table", ex );
            }
        }  // end AddTableType

        public void SetKineticaSourceClassToTypeMapping( Type? objectType, KineticaType kineticaType )
        {
            if ( objectType != null )
                this.kineticaTypeLookup[objectType] = kineticaType;
            return;
        }  // end SetKineticaSourceClassToTypeMapping



        public void DecodeRawBinaryDataUsingRecordType<T>( KineticaType record_type,
                                                           IList<byte[]> records_binary,
                                                           IList<T> records ) where T : new()
        {
            // Using the KineticaType object, decode all the records from avro binary encoding
            foreach ( var bin_record in records_binary )
            {
                T obj = AvroDecode<T>( bin_record, record_type );
                records.Add( obj );
            }
        }  // DecodeRawBinaryDataUsingRecordType


        public void DecodeRawBinaryDataUsingSchemaString<T>( string schema_string,
                                                             IList<byte[]> records_binary,
                                                             IList<T> records ) where T : new()
        {
            // Create a KineticaType object based on the schema string
            KineticaType ktype = new("", schema_string, null);

            // Using the KineticaType object, decode all the records from avro binary encoding
            foreach ( var bin_record in records_binary )
            {
                T obj = AvroDecode<T>( bin_record, ktype );
                records.Add( obj );
            }
        }  // DecodeRawBinaryDataUsingSchemaString

        public void DecodeRawBinaryDataUsingSchemaString<T>( IList<string> schema_strings,
                                                             IList<IList<byte[]>> lists_records_binary,
                                                             IList<IList<T>> record_lists ) where T : new()
        {
            // Check that the list of schemas and list of binary encode data match in length
            if ( schema_strings.Count != lists_records_binary.Count )
                throw new KineticaException( "List of schemas and list of binary encoded data do not match in count." );

            // Using the KineticaType object, decode all the records from avro binary encoding
            for ( int i = 0; i < schema_strings.Count; ++i )
            {
                // Create a KineticaType object based on the schema string
                KineticaType ktype = new( "", schema_strings[ i ], null );

                // Get the binary encoded data for this list
                IList<byte[]> records_binary = lists_records_binary[ i ];

                // Create a container to put the decoded records
                IList<T> records = [];

                // The inner list actually contains the binary data
                foreach ( var bin_record in records_binary )
                {
                    T obj = AvroDecode<T>( bin_record, ktype );
                    records.Add( obj );
                }
                // Add the records into the outgoing list
                record_lists.Add( records );
            }
        }  // DecodeRawBinaryDataUsingSchemaString


        public void DecodeRawBinaryDataUsingTypeIDs<T>( IList<string> type_ids,
                                                        IList<byte[]> records_binary,
                                                        IList<T> records ) where T : new()
        {
            // Make sure that the length of the type IDs and records are the same
            if ( type_ids.Count != records_binary.Count )
                throw new KineticaException( "Unequal numbers of type IDs and binary encoded data objects provided." );

            // Decode all the records
            for ( int i = 0; i < records_binary.Count; ++i )
            {
                // Per object, use the respective type ID to create the appropriate KineticaType
                KineticaType ktype = KineticaType.fromTypeID( this, type_ids[ i ] );

                // Using the KineticaType object, decode the record.
                T obj = AvroDecode<T>( records_binary[ i ], ktype );
                records.Add( obj );
            }
        }  // DecodeRawBinaryDataUsingTypeIDs


        public void DecodeRawBinaryDataUsingTypeIDs<T>( IList<string> type_ids,
                                                        IList<IList<byte[]>> lists_records_binary,
                                                        IList<IList<T>> record_lists ) where T : new()
        {
            // Make sure that the length of the type IDs and records are the same
            if ( type_ids.Count != lists_records_binary.Count )
                throw new KineticaException( "Unequal numbers of type IDs and binary encoded data objects provided." );

            // Decode all the records
            for ( int i = 0; i < lists_records_binary.Count; ++i )
            {
                // Per object, use the respective type ID to create the appropriate KineticaType
                KineticaType ktype = KineticaType.fromTypeID( this, type_ids[ i ] );

                // Get the binary encoded data for this list
                IList<byte[]> records_binary = lists_records_binary[ i ];

                // Create a container to put the decoded records
                IList<T> records = [];

                // The inner list actually contains the binary data
                foreach ( var bin_record in records_binary )
                {
                    // Using the KineticaType object, decode the record.
                    T obj = AvroDecode<T>( bin_record, ktype );
                    records.Add( obj );
                }
                // Add the records into the outgoing list
                record_lists.Add( records );
            }
        }  // DecodeRawBinaryDataUsingTypeIDs

        #region Request Submission API (Matches Rust gpudb.rs design)

        private TResponse SubmitRequest<TResponse>(string endpoint, object request, bool enableCompression = false, bool avroEncoding = true) where TResponse : new()
        {
            // Encode the request
            byte[] requestBytes = avroEncoding
                ? AvroEncode(request)
                : Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(request));

            // If HA failover is not available, just submit the request directly
            if (_haFailoverManager == null || _haFailoverManager.HARingSize <= 1)
            {
                string fullUrl = Url + endpoint;
                RawKineticaResponse kineticaResponse = SubmitRequestToUrlInternal(fullUrl, requestBytes, enableCompression, avroEncoding);
                return DecodeResponse<TResponse>(kineticaResponse, avroEncoding);
            }

            // HA failover is available - attempt with failover logic
            var currentUrl = _haFailoverManager.GetUrl();
            if (currentUrl == null)
            {
                throw new KineticaException("No URL available");
            }

            var originalUrl = currentUrl;
            int currentSwitchCount = _haFailoverManager.NumClusterSwitches;

            while (true)
            {
                try
                {
                    // Build the full URL with the endpoint
                    string fullUrl = currentUrl.ToString().TrimEnd('/') + endpoint;
                    RawKineticaResponse kineticaResponse = SubmitRequestToUrlInternal(fullUrl, requestBytes, enableCompression, avroEncoding);
                    return DecodeResponse<TResponse>(kineticaResponse, avroEncoding);
                }
                catch (Exception ex) when (IsConnectionError(ex))
                {
                    // This is a connection error - attempt failover
                    try
                    {
                        currentUrl = _haFailoverManager.SwitchUrl(originalUrl, currentSwitchCount, IsKineticaRunning);
                        // Update the main URL reference
                        Url = currentUrl.ToString().TrimEnd('/');
                        URL = currentUrl;
                    }
                    catch (KineticaException)
                    {
                        // Failover failed - re-throw the original exception
                        throw new KineticaException($"Connection failed and HA failover unsuccessful: {ex.Message}", ex);
                    }
                }
                catch (KineticaException)
                {
                    // API error from server - don't failover, just rethrow
                    throw;
                }
            }
        }

        private async System.Threading.Tasks.Task<TResponse> SubmitRequestAsync<TResponse>(
            string endpoint,
            object request,
            bool enableCompression = false,
            bool avroEncoding = true,
            System.Threading.CancellationToken cancellationToken = default)
            where TResponse : new()
        {
            // Encode the request
            byte[] requestBytes = avroEncoding
                ? AvroEncode(request)
                : Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(request));

            // If HA failover is not available, just submit the request directly
            if (_haFailoverManager == null || _haFailoverManager.HARingSize <= 1)
            {
                string fullUrl = Url + endpoint;
                RawKineticaResponse kineticaResponse = await SubmitRequestToUrlInternalAsync(
                    fullUrl, requestBytes, enableCompression, avroEncoding, cancellationToken);
                return DecodeResponse<TResponse>(kineticaResponse, avroEncoding);
            }

            // HA failover is available - attempt with failover logic
            var currentUrl = _haFailoverManager.GetUrl();
            if (currentUrl == null)
            {
                throw new KineticaException("No URL available");
            }

            var originalUrl = currentUrl;
            int currentSwitchCount = _haFailoverManager.NumClusterSwitches;

            while (true)
            {
                try
                {
                    // Build the full URL with the endpoint
                    string fullUrl = currentUrl.ToString().TrimEnd('/') + endpoint;
                    RawKineticaResponse kineticaResponse = await SubmitRequestToUrlInternalAsync(
                        fullUrl, requestBytes, enableCompression, avroEncoding, cancellationToken);
                    return DecodeResponse<TResponse>(kineticaResponse, avroEncoding);
                }
                catch (Exception ex) when (IsConnectionError(ex))
                {
                    // This is a connection error - attempt failover
                    try
                    {
                        currentUrl = _haFailoverManager.SwitchUrl(originalUrl, currentSwitchCount, IsKineticaRunning);
                        // Update the main URL reference
                        Url = currentUrl.ToString().TrimEnd('/');
                        URL = currentUrl;
                    }
                    catch (KineticaException)
                    {
                        // Failover failed - re-throw the original exception
                        throw new KineticaException($"Connection failed and HA failover unsuccessful: {ex.Message}", ex);
                    }
                }
                catch (KineticaException)
                {
                    // API error from server - don't failover, just rethrow
                    throw;
                }
            }
        }

        public TResponse SubmitRequestRaw<TResponse>(Uri url, object request, bool enableCompression = false, bool avroEncoding = true) where TResponse : new()
        {
            // Encode the request
            byte[] requestBytes = avroEncoding
                ? AvroEncode(request)
                : Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(request));

            RawKineticaResponse kineticaResponse = SubmitRequestToUrlInternal(url.ToString(), requestBytes, enableCompression, avroEncoding);
            return DecodeResponse<TResponse>(kineticaResponse, avroEncoding);
        }

        public RawKineticaResponse SubmitRequestRawBytes(Uri url, byte[] requestBytes)
        {
            return SubmitRequestToUrlInternal(url.ToString(), requestBytes, UseSnappy, true);
        }

        #endregion

        #region Internal HTTP Helpers

        private TResponse DecodeResponse<TResponse>(RawKineticaResponse kineticaResponse, bool avroEncoding) where TResponse : new()
        {
            if (avroEncoding)
            {
                return AvroDecode<TResponse>(kineticaResponse.data);
            }
            else // JSON
            {
                kineticaResponse.data_str = kineticaResponse.data_str.Replace("\\U", "\\u");
                return JsonConvert.DeserializeObject<TResponse>(kineticaResponse.data_str);
            }
        }

        public bool IsKineticaRunning(Uri url)
        {
            try
            {
                // Simple GET request to check if server is running
                // We can't use the transport layer here as it only supports POST
                // So we'll create a temporary HttpClient for this check
                using var client = new HttpClient { Timeout = TimeSpan.FromSeconds(5) };
                using var response = client.GetAsync(url).Result;

                if (response.IsSuccessStatusCode)
                {
                    string responseText = response.Content.ReadAsStringAsync().Result;
                    return responseText.Contains("Kinetica is running!");
                }
                return false;
            }
            catch
            {
                return false;
            }
        }

        private RawKineticaResponse SubmitRequestToUrlInternal(string url, byte[] requestBytes, bool enableCompression, bool avroEncoding)
        {
            try
            {
                // Apply Snappy compression if enabled
                byte[] bodyBytes;
                string contentType;

                if (enableCompression && avroEncoding)
                {
                    // Compress using Snappier (pure managed Snappy implementation)
                    bodyBytes = Snappy.CompressToArray(requestBytes);
                    contentType = "application/x-snappy";
                }
                else
                {
                    bodyBytes = requestBytes;
                    contentType = avroEncoding ? "application/octet-stream" : "application/json";
                }

                // Use the HTTP transport layer to send the request
                var responseBytes = _transport.Post(
                    url,
                    bodyBytes,
                    contentType,
                    Authorization,
                    System.Threading.CancellationToken.None);

                // Decode the response
                if (avroEncoding)
                {
                    return AvroDecode<RawKineticaResponse>(responseBytes);
                }
                else // JSON
                {
                    var responseString = Encoding.UTF8.GetString(responseBytes);
                    responseString = responseString.Replace("\\U", "\\u");
                    return JsonConvert.DeserializeObject<RawKineticaResponse>(responseString)
                           ?? throw new KineticaException("Failed to deserialize response");
                }
            }
            catch (KineticaTransportException tex)
            {
                // HTTP transport returned a non-2xx status code
                // The server may have encoded an error message in the response body
                try
                {
                    RawKineticaResponse? serverResponse;
                    if (avroEncoding)
                    {
                        serverResponse = AvroDecode<RawKineticaResponse>(tex.Body);
                    }
                    else // JSON
                    {
                        var responseString = Encoding.UTF8.GetString(tex.Body);
                        serverResponse = JsonConvert.DeserializeObject<RawKineticaResponse>(responseString);
                    }

                    throw new KineticaException(
                        serverResponse?.message ?? $"Server returned HTTP {tex.StatusCode}",
                        tex.StatusCode,
                        tex);
                }
                catch (KineticaException)
                {
                    throw;
                }
                catch
                {
                    // Could not decode error response - throw with status code
                    throw new KineticaException($"Server returned HTTP {tex.StatusCode}", tex.StatusCode, tex);
                }
            }
            catch (HttpRequestException ex)
            {
                throw new KineticaException(ex.ToString(), ex);
            }
            catch (TaskCanceledException ex)
            {
                throw new KineticaException("Request timed out: " + ex.ToString(), ex);
            }
            catch (OperationCanceledException ex)
            {
                throw new KineticaException("Request cancelled: " + ex.ToString(), ex);
            }
            catch (KineticaException)
            {
                throw;
            }
            catch (Exception ex)
            {
                throw new KineticaException(ex.ToString(), ex);
            }
        }

        private async Task<RawKineticaResponse> SubmitRequestToUrlInternalAsync(
            string url,
            byte[] requestBytes,
            bool enableCompression,
            bool avroEncoding,
            System.Threading.CancellationToken cancellationToken = default)
        {
            try
            {
                // Apply Snappy compression if enabled
                byte[] bodyBytes;
                string contentType;

                if (enableCompression && avroEncoding)
                {
                    // Compress using Snappier (pure managed Snappy implementation)
                    bodyBytes = Snappy.CompressToArray(requestBytes);
                    contentType = "application/x-snappy";
                }
                else
                {
                    bodyBytes = requestBytes;
                    contentType = avroEncoding ? "application/octet-stream" : "application/json";
                }

                // Use the HTTP transport layer to send the request asynchronously
                var responseBytes = await _transport
                    .PostAsync(url, bodyBytes, contentType, Authorization, cancellationToken)
                    .ConfigureAwait(false);

                // Decode the response
                if (avroEncoding)
                {
                    return AvroDecode<RawKineticaResponse>(responseBytes);
                }
                else // JSON
                {
                    var responseString = Encoding.UTF8.GetString(responseBytes);
                    responseString = responseString.Replace("\\U", "\\u");
                    return JsonConvert.DeserializeObject<RawKineticaResponse>(responseString)
                           ?? throw new KineticaException("Failed to deserialize response");
                }
            }
            catch (KineticaTransportException tex)
            {
                // HTTP transport returned a non-2xx status code
                // The server may have encoded an error message in the response body
                try
                {
                    RawKineticaResponse? serverResponse;
                    if (avroEncoding)
                    {
                        serverResponse = AvroDecode<RawKineticaResponse>(tex.Body);
                    }
                    else // JSON
                    {
                        var responseString = Encoding.UTF8.GetString(tex.Body);
                        serverResponse = JsonConvert.DeserializeObject<RawKineticaResponse>(responseString);
                    }

                    throw new KineticaException(
                        serverResponse?.message ?? $"Server returned HTTP {tex.StatusCode}",
                        tex.StatusCode,
                        tex);
                }
                catch (KineticaException)
                {
                    throw;
                }
                catch
                {
                    // Could not decode error response - throw with status code
                    throw new KineticaException($"Server returned HTTP {tex.StatusCode}", tex.StatusCode, tex);
                }
            }
            catch (HttpRequestException ex)
            {
                throw new KineticaException(ex.ToString(), ex);
            }
            catch (TaskCanceledException ex)
            {
                throw new KineticaException("Request timed out: " + ex.ToString(), ex);
            }
            catch (OperationCanceledException ex)
            {
                throw new KineticaException("Request cancelled: " + ex.ToString(), ex);
            }
            catch (KineticaException)
            {
                throw;
            }
            catch (Exception ex)
            {
                throw new KineticaException(ex.ToString(), ex);
            }
        }

        public static bool IsConnectionError(Exception ex)
        {
            return ex is System.Net.WebException webEx && webEx.Status != WebExceptionStatus.ProtocolError ||
                   ex is System.Net.Sockets.SocketException ||
                   ex is IOException ||
                   ex is System.Net.Http.HttpRequestException ||
                   ex is TaskCanceledException ||
                   (ex is KineticaException kex && kex.Message.Contains("connection", StringComparison.OrdinalIgnoreCase));
        }

        internal Uri? ForceHAFailover(Uri currentUrl, int currentSwitchCount)
        {
            if (_haFailoverManager == null || _haFailoverManager.HARingSize <= 1)
            {
                return null;
            }

            try
            {
                var newUrl = _haFailoverManager.SwitchUrl(currentUrl, currentSwitchCount, IsKineticaRunning);
                Url = newUrl.ToString().TrimEnd('/');
                URL = newUrl;
                return newUrl;
            }
            catch (KineticaException)
            {
                return null;
            }
        }

        public int HARingSize => _haFailoverManager?.HARingSize ?? 1;

        #endregion

        #region Type Registration and Encoding

        private void SetDecoderIfMissing(string typeId, string label, string schemaString, IDictionary<string, IList<string>> properties)
        {
            // If the table is a collection, it does not have a proper type so ignore it

            if (typeId == "<collection>")
            {
                return;
            }

            knownTypes.GetOrAdd(typeId, (s) =>
            {
                return new KineticaType(label, schemaString, properties);
            });
            typeNameLookup[label] = typeId;
        }


        private KineticaType? GetType(string typeName)
        {
            KineticaType? type = null;
            if (typeNameLookup.TryGetValue(typeName, out string? typeId))
            {
                knownTypes.TryGetValue(typeId, out type);
            }

            return type;
        }


        private KineticaType? LookupKineticaType( Type objectType )
        {
            if (!kineticaTypeLookup.TryGetValue(objectType, out KineticaType? value))
                return null; // none found

            return value;
        }  // LookupKineticaType()


        internal byte[] AvroEncode(object obj)
        {
            // Create a stream that will allow us to view the underlying memory
            using ( var ms = new MemoryStream())
            {
                // Write the object to the memory stream
                // If obj is an ISpecificRecord, this is more efficient
                if ( obj is Avro.Specific.ISpecificRecord)
                {
                    var schema = (obj as Avro.Specific.ISpecificRecord).Schema;
                    Avro.Specific.SpecificDefaultWriter writer = new(schema);
                    writer.Write(schema, obj, new BinaryEncoder(ms));
                }
                else // Not an ISpecificRecord - this way is less efficient
                {
                    // Get the KineticaType associated with the object to be encoded
                    Type obj_type = obj.GetType();
                    KineticaType? ktype = LookupKineticaType( obj_type );
                    if ( ktype == null )
                    {
                        throw new KineticaException( "No known KineticaType associated with the given object.  " +
                                                     "Need a known KineticaType to encode the object." );
                    }

                    // Make a copy of the object to send as a GenericRecord, then write that to the memory stream
                    var schema = KineticaData.SchemaFromType( obj.GetType(), ktype );
                    var recordToSend = MakeGenericRecord( obj, ktype );
                    var writer = new Avro.Generic.DefaultWriter(schema);
                    writer.Write(schema, recordToSend, new BinaryEncoder(ms));
                }

                // Get the memory from the stream
                return ms.ToArray();
            }
        }  // end AvroEncode

        private Avro.Generic.GenericRecord MakeGenericRecord( object obj, KineticaType ktype )
        {
            // Get the schema
            var schema = KineticaData.SchemaFromType( obj.GetType(), ktype );

            // Create a new GenericRecord for this schema
            var recordToSend = new Avro.Generic.GenericRecord(schema);

            // Copy each field from obj to recordToSend
            foreach ( var field in schema.Fields)
            {
                var property = obj.GetType()
                                .GetProperties()
                                .FirstOrDefault(prop => prop.Name.ToLowerInvariant() == field.Name.ToLowerInvariant());

                if (property == null) continue;

                recordToSend.Add(field.Name, property.GetValue(obj, null));
            }

            // Return the newly created object
            return recordToSend;
        }

        internal T AvroDecode<T>(byte[] bytes, KineticaType? ktype = null) where T : new()
        {
            // Get the schema
            var schema = KineticaData.SchemaFromType( typeof(T), ktype );

            // Create a stream to read the binary data
            using (var ms = new MemoryStream(bytes))
            {
                // Create a new object to return
                T obj = new();
                if (obj is Avro.Specific.ISpecificRecord)
                {
                    var reader = new Avro.Specific.SpecificDefaultReader(schema, schema);
                    reader.Read(obj, new BinaryDecoder(ms));
                }
                else
                {
                    // Not ISpecificRecord, so first read into a new GenericRecord
                    var reader = new Avro.Generic.DefaultReader(schema, schema);
                    Avro.Generic.GenericRecord recordToReceive = new(schema);
                    reader.Read(recordToReceive, new BinaryDecoder(ms));

                    // Now, copy all the fields from the GenericRecord to obj
                    foreach (var field in schema.Fields)
                    {
                        var property = obj.GetType()
                                        .GetProperties()
                                        .FirstOrDefault(prop => prop.Name.ToLowerInvariant() == field.Name.ToLowerInvariant());

                        if (property == null) continue;

                        // Try to get the property
                        if (recordToReceive.TryGetValue(field.Name, out object val))
                        {
                            // If successful, write the property to obj
                            property.SetValue(obj, val);
                        }
                    }  // end foreach
                }  // end if-else

                // Return the new object
                return obj;
            }  // end using
        }  // end AvroDecode<T>


        internal T AvroDecode<T>(Stream stream) where T : Avro.Specific.ISpecificRecord, new()
        {
            // T obj = new T(); // Activator.CreateInstance<T>();
            var schema = KineticaData.SchemaFromType( typeof(T), null );
            var reader = new Avro.Specific.SpecificReader<T>(schema, schema);
            return reader.Read(default, new BinaryDecoder(stream));
        }

        #endregion
    }  // end class Kinetica