Kinetica.cs

Go to the documentation of this file.

using Avro.IO;
using Newtonsoft.Json;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Text;


namespace kinetica
{
    public partial class Kinetica
    {
        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 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 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 object class type to KineticaType lookup table
        private Dictionary<Type, KineticaType> kineticaTypeLookup = [];

        public Kinetica( string url_str, Options? options = null )
        {
            Url = url_str;
            URL = new Uri( url_str );
            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;
                // TODO: executor?
            }
        }

        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 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.Add( 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


        internal TResponse SubmitRequest<TResponse>( Uri url, object request, bool enableCompression = false, bool avroEncoding = true ) where TResponse : new()
        {
            // Get the bytes to send, encoded in the requested way
            byte[] requestBytes;
            if ( avroEncoding )
            {
                requestBytes = AvroEncode( request );
            }
            else // JSON
            {
                string str = JsonConvert.SerializeObject(request);
                requestBytes = Encoding.UTF8.GetBytes( str );
            }

            // Send request, and receive response
            RawKineticaResponse kineticaResponse = SubmitRequestRaw( url.ToString(), requestBytes, enableCompression, avroEncoding, false);

            // Decode response payload
            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 );
            }
        }  // end SubmitRequest( URL )


        private TResponse SubmitRequest<TResponse>(string endpoint, object request, bool enableCompression = false, bool avroEncoding = true) where TResponse : new()
        {
            // Get the bytes to send, encoded in the requested way
            byte[] requestBytes;
            if (avroEncoding)
            {
                requestBytes = AvroEncode(request);
            }
            else // JSON
            {
                string str = JsonConvert.SerializeObject(request);
                requestBytes = Encoding.UTF8.GetBytes(str);
            }

            // Send request, and receive response
            RawKineticaResponse kineticaResponse = SubmitRequestRaw(endpoint, requestBytes, enableCompression, avroEncoding);

            // Decode response payload
            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);
            }
        }  // end SubmitRequest( endpoint )



        private RawKineticaResponse? SubmitRequestRaw(string url, byte[] requestBytes, bool enableCompression, bool avroEncoding, bool only_endpoint_given = true)
        {
            try
            {
                if ( only_endpoint_given )
                    url = (Url + url);
                var request = (HttpWebRequest)WebRequest.Create( url );
                request.Method = "POST";
                //request.UseDefaultCredentials = true;
                request.ContentType = avroEncoding ? "application/octet-stream" : "application/json";
                request.ContentLength = requestBytes.Length;

                // Handle the authorization
                if ( this.Authorization != null )
                {
                    request.Headers.Add( "Authorization", Authorization );
                }


                // Write the binary request data
                using ( var dataStream = request.GetRequestStream())
                {
                    dataStream.Write(requestBytes, 0, requestBytes.Length);
                }

                // Send to the server and await a response
                using (var response = (HttpWebResponse)request.GetResponse())
                {
                    // Parse the response
                    if (response.StatusCode == HttpStatusCode.OK)
                    {
                        using (var responseStream = response.GetResponseStream())
                        {
                            if (avroEncoding)
                            {
                                return AvroDecode<RawKineticaResponse>(responseStream);
                            }
                            else // JSON
                            {
                                using (StreamReader reader = new(responseStream, Encoding.UTF8))
                                {
                                    var responseString = reader.ReadToEnd();
                                    return JsonConvert.DeserializeObject<RawKineticaResponse>(responseString);
                                }
                            }
                        }
                    }
                }
            }
            catch (System.Net.WebException ex)
            {
                // Skip trying parsing the message if not a protocol error
                if ( ex.Status != WebExceptionStatus.ProtocolError )
                    throw new KineticaException( ex.ToString(), ex );

                // Get the error message from the server response
                var response = ex.Response;
                var responseStream = response.GetResponseStream();
                string responseString;
                RawKineticaResponse serverResponse;
                // Decode the response packet
                if (avroEncoding)
                {
                    serverResponse = AvroDecode<RawKineticaResponse>(responseStream);
                }
                else // JSON
                {
                    using (StreamReader reader = new(responseStream, Encoding.UTF8))
                    {
                        responseString = reader.ReadToEnd();
                        serverResponse = JsonConvert.DeserializeObject<RawKineticaResponse>(responseString);
                    }
                }
                // Throw the error message found within the response packet
                throw new KineticaException( serverResponse.message );
            }
            catch (Exception ex)
            {
                throw new KineticaException(ex.ToString(), ex);
            }

            return null;
        }

        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;
        }

        private 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>


        private 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));
        }
        /*
        private T AvroDecode<T>(string str) where T : new()
        {
            return AvroDecode<T>(Encoding.UTF8.GetBytes(str));
        }
        */
    }  // end class Kinetica
}  // end namespace kinetica