KineticaType.cs

Go to the documentation of this file.

using System;
using System.Collections.Generic;
using System.Linq;
using Avro;
using Newtonsoft.Json.Linq;

namespace kinetica
{
    public class KineticaType
    {
        public class Column
        {
            public enum ColumnType
            {
                BYTES = Avro.EnumSchema.Type.Bytes,
                DOUBLE = Avro.EnumSchema.Type.Double,
                FLOAT = Avro.EnumSchema.Type.Float,
                INT = Avro.EnumSchema.Type.Int,
                LONG = Avro.EnumSchema.Type.Long,
                STRING = Avro.EnumSchema.Type.String,
                DEFAULT = Avro.EnumSchema.Type.Error
            };

            private string m_name;
            private ColumnType m_type;
            private bool m_isNullable;
            private IList<string> m_properties;

            public Column(string name, ColumnType type, IList<string>? properties = null)
            {
                m_name = name;
                m_type = type;
                m_isNullable = false;
                m_properties = properties ?? [];

                Initialize();
            }

            public string getName() { return m_name; }

            public ColumnType getType() { return m_type; }

            public bool isNullable() { return m_isNullable; }

            public IList<string> getProperties() { return m_properties; }

            internal void setIsNullable( bool val ) { m_isNullable = val;  }

            public string getTypeString()
            {
                return m_type switch
                {
                    ColumnType.BYTES => "bytes",
                    ColumnType.DOUBLE => "double",
                    ColumnType.FLOAT => "float",
                    ColumnType.INT => "int",
                    ColumnType.LONG => "long",
                    ColumnType.STRING => "string",
                    _ => throw new KineticaException("Unsupported column type: " + m_type),
                };
            }  // end getTypeString()

            private void Initialize()
            {
                if (string.IsNullOrEmpty(m_name))
                {
                    throw new ArgumentException("Name must not be empty.");
                }

                switch (m_type)
                {
                    case ColumnType.BYTES:
                    case ColumnType.DOUBLE:
                    case ColumnType.FLOAT:
                    case ColumnType.INT:
                    case ColumnType.LONG:
                    case ColumnType.STRING:
                        break;

                    default:
                        throw new ArgumentException($"Column {m_name} must be of type BYTES, DOUBLE, FLOAT, INT, LONG or STRING.");
                }

                foreach (var it in m_properties)
                {
                    if (string.IsNullOrEmpty(it))
                    {
                        throw new ArgumentException("Properties must not be empty.");
                    }

                    if (!m_isNullable && (it == ColumnProperty.NULLABLE))
                    {
                        m_isNullable = true;
                    }
                }
            }

            public override string ToString()
            {
                return $"{m_name} ({m_type})";
            }
        }  // end class Column

        private class TypeData
        {
            public string? label;
            public IList<Column> columns = [];
            public Dictionary<string, int> columnMap = [];
            public string? schemaString = null;
            public Schema? schema = null;
            public Type? sourceType = null;
        }

        private TypeData m_data = new();
        private IDictionary<string, IList<string>> m_properties = new Dictionary<string, IList<string>>();
        private string? m_typeId = null;

        public static KineticaType fromTable(Kinetica kinetica, string tableName)
        {
            var response = kinetica.showTable(tableName);
            var typeIdCount = response.type_ids.Count;

            if (typeIdCount == 0)
            {
                throw new KineticaException($"Table {tableName} does not exist.");
            }

            string typeId = response.type_ids[0];
            if (typeIdCount > 1)
            {
                for (int i = 1; i < typeIdCount; ++i)
                {
                    if (response.type_ids[i] != typeId)
                    {
                        throw new KineticaException("Table {tableName} is not homogeneous.");
                    }
                }
            }

            return new KineticaType(response.type_labels[0], response.type_schemas[0], response.properties[0], typeId );
        }

        public static KineticaType fromTypeID(Kinetica kinetica, string typeId)
        {
            var response = kinetica.showTypes(typeId, "");

            if (response.type_ids.Count < 1)
            {
                throw new KineticaException($"Type {typeId} does not exist.");
            }

            return new KineticaType(response.labels[0], response.type_schemas[0], response.properties[0]);
        }


        public static KineticaType fromDynamicSchema( string dynamicTableSchemaString,
                                                      Object[] columnHeaders, Object[] columnTypes )
        {
            // Make sure that the lists of column names and types are of the same length
            if ( columnHeaders.Length != columnTypes.Length )
                throw new KineticaException($"List of column names ({columnHeaders.Length}) and types ({columnTypes.Length}) are not of the same length." );

            // Parse the schema string so that we can later check if a given column is nullable
            JObject dynamicSchemaJson;
            try
            {
                dynamicSchemaJson = JObject.Parse( dynamicTableSchemaString );
            }
            catch ( Exception ex )
            {
                throw new KineticaException( ex.ToString() );
            }

            // Create a delegate for checking if a field/column is nullable
            // ------------------------------------------------------------
            // The first parameter is the column name, the second is the JSON object
            var isColumnNullable = new Func<string, JObject, bool>( (columnName, schemaJson) => {
                // Find the appropriate field
                bool foundField = false;
                JToken? fieldList = schemaJson["fields"];
                if (fieldList != null)
                {
                    foreach (var field in fieldList)
                    {
                        if ((string?)field["name"] == columnName)
                        {
                            foundField = true; // found it!
                            // Get the type and see if it's a nullable type
                            // (each field is an array of the column type; so need
                            // to extract the type element first)
                            JToken? fieldType = field["type"];
                            if (fieldType != null)
                            {
                                var typeElement = fieldType["items"];
                                if (typeElement == null || typeElement is JValue) // not an array, so can't be nullable
                                    return false;
                                // If the type is an array and the second value is 'null', then it's a nullable
                                if ((typeElement is JArray) && ((string)(typeElement as JArray)[1] == "null"))
                                    return true;
                                return false;
                            }
                        } // end if
                    }  // end foreach
                }
                if ( !foundField )
                    throw new KineticaException( $"Could not find the field named '{columnName}'" );
                return false; // shouldn't ever get here
            } );

            // Create appropriate columns and column properties
            // ------------------------------------------------
            List<Column> columns = [];
            Dictionary<string, IList<string>> columnProperties = [];
            for ( int i = 0; i < columnHeaders.Length; ++i )
            {
                // Get the column's name
                string column_name = (string)columnHeaders[i];

                // Get the column's type in string format, which might be a property and not a primitive type
                // (if so, then we'll have to infer the primitive type and save the property)
                string columnTypeString = (string)columnTypes[i];

                // Need to create a list for the properties of the column (we'll
                // extract at most one from the column type)
                List<string> columnProperty = [];

                // We need to also infer the primitive type for this column
                Column.ColumnType columnType;

                // Infer the type and property from the given 'type'
                switch ( columnTypeString )
                {
                    // Primitive type string (and all properties based on it)
                    case "string":
                        columnType = Column.ColumnType.STRING;
                        break;

                    // All properties allowed for the primitive string type
                    case ColumnProperty.CHAR1:
                    case ColumnProperty.CHAR2:
                    case ColumnProperty.CHAR4:
                    case ColumnProperty.CHAR8:
                    case ColumnProperty.CHAR16:
                    case ColumnProperty.CHAR32:
                    case ColumnProperty.CHAR64:
                    case ColumnProperty.CHAR128:
                    case ColumnProperty.CHAR256:
                    case ColumnProperty.DATE:
                    case ColumnProperty.DATETIME:
                    case ColumnProperty.DECIMAL: 
                    case ColumnProperty.IPV4:
                    case ColumnProperty.TIME:
                        columnType = Column.ColumnType.STRING;
                        columnProperty.Add( columnTypeString );
                        break;

                    // Primitive type integer
                    case "int":
                        columnType = Column.ColumnType.INT;
                        break;

                    // Properties allowed for the primitive integer type
                    case ColumnProperty.INT8:
                    case ColumnProperty.INT16:
                        columnType = Column.ColumnType.INT;
                        columnProperty.Add( columnTypeString );
                        break;

                    // Primitive type long
                    case "long":
                        columnType = Column.ColumnType.LONG;
                        break;

                    // Properties allowed for the long type
                    case ColumnProperty.TIMESTAMP:
                        columnType = Column.ColumnType.LONG;
                        columnProperty.Add( columnTypeString );
                        break;

                    // Primitive type float
                    case "float":
                        columnType = Column.ColumnType.FLOAT;
                        break;

                    // Primitive type double
                    case "double":
                        columnType = Column.ColumnType.DOUBLE;
                        break;

                    // Primitive type bytes
                    case "bytes":
                        columnType = Column.ColumnType.BYTES;
                        break;

                    default:
                        throw new KineticaException($"Unknown data type/property: {columnTypeString}");
                }  // end switch

                // Check if the column is nullable (where the column name is "column_#" as returned by Kinetica)
                if ( isColumnNullable( $"column_{i + 1}", dynamicSchemaJson ) )
                    columnProperty.Add( ColumnProperty.NULLABLE );

                // Now that we have the name, the type and potentially a property for the column,
                // create a Column type and add it to the list
                Column column = new( column_name, columnType, columnProperty );
                columns.Add( column );

                // Also, save the column property in the column name->property map
                columnProperties.Add( column_name, columnProperty );
            }  // end looping over column headers and types


            // Create and return the KineticaType object based on the columns and properties
            return new KineticaType( "", columns, columnProperties );
        }  // end fromDynamicSchema()



        public static KineticaType fromClass( Type recordClass, IDictionary<string, IList<string>> properties = null )
        {
            return fromClass( recordClass, "", properties );
        }   // end fromClass()


        public static KineticaType fromClass( Type recordClass, string label, IDictionary<string, IList<string>>? properties = null )
        {
            // Get the fields in order (******skipping properties inherited from base classes******)
            // (fields only from this type, i.e. do not include any inherited fields), and public types only
            System.Reflection.PropertyInfo[] typeProperties = recordClass.GetProperties( System.Reflection.BindingFlags.DeclaredOnly |
                                                                                          System.Reflection.BindingFlags.Instance |
                                                                                          System.Reflection.BindingFlags.Public );

            Array.Sort( typeProperties, (p1, p2) => p1.MetadataToken.CompareTo( p2.MetadataToken ) );

            // Need to have a list of columns
            List<Column> columns = [];
            List<string> columnNames = [];

            // Per property, check that it is one of: int, long, float, double, string, bytes
            foreach ( var typeProperty in typeProperties )
            {
                string            columnName = "";
                Column.ColumnType columnType = Column.ColumnType.DEFAULT;
                IList<string>?    columnProperties = null;
                bool              isColumnNullable = false;

                // Get the column name
                columnName = typeProperty.Name;

                Type? propertyType = typeProperty.PropertyType;

                // Check if the field is nullable (declared as T? or Nullable<T>)
                if ( typeProperty.PropertyType.IsGenericType &&
                          ( typeProperty.PropertyType.GetGenericTypeDefinition() == typeof( Nullable<> ) ) )
                {   // the field is a nullable field
                    isColumnNullable = true;
                    // Change the property type to be the underlying type
                    propertyType = Nullable.GetUnderlyingType(propertyType);
                }

                // Check the column data type (must be one of int, long, float, double, string, and bytes)
                if ( propertyType == typeof( System.String ) )
                {
                    columnType = Column.ColumnType.STRING;
                }
                else if ( propertyType == typeof( System.Int32 ) )
                {
                    columnType = Column.ColumnType.INT;
                }
                else if ( propertyType == typeof( System.Int64 ) )
                {
                    columnType = Column.ColumnType.LONG;
                }
                else if ( propertyType == typeof( float ) )
                {
                    columnType = Column.ColumnType.FLOAT;
                }
                else if ( propertyType == typeof( double ) )
                {
                    columnType = Column.ColumnType.DOUBLE;
                }
                else if ( propertyType == typeof( byte ) )
                {
                    columnType = Column.ColumnType.BYTES;
                }
                else
                    throw new KineticaException( "Unsupported data type for " + propertyType?.Name +
                                              ": " + propertyType +
                                              " (must be one of int, long, float, double, string, and byte)" );

                // Extract the given column's properties, if any
                properties?.TryGetValue(columnName, out columnProperties);

                // Keep a list of the column names for checking the properties
                columnNames.Add( columnName );

                // Create the column
                Column column = new( columnName, columnType, columnProperties );
                if ( isColumnNullable ) // Set the appropriate nullable flag for the column
                    column.setIsNullable( true );

                // Save the column
                columns.Add( column );
            }  // end looping over all members of the class type

            // Check for extraneous properties
            if (properties != null)
            {
                IEnumerable<string> propertyKeys = properties.Keys;
                var unknownColumns = propertyKeys.Where(e => !columnNames.Contains(e));
                // Check if any property is provided for wrong/non-existing columns
                if (unknownColumns.Any())
                    throw new KineticaException("Properties specified for unknown columns.");
            }
            else properties = new Dictionary<string, IList<string>>();

            // Create the kinetica type
            KineticaType kType = new(label, columns, properties);

            // Save the class information in the type
            kType.saveSourceType( recordClass );

            return kType;
        } // end fromClass()


        public static KineticaType fromObject( Object recordObj, IDictionary<string, IList<string>> properties = null )
        {
            return fromObject( recordObj, "", properties );
        }  // end fromObject()


        public static KineticaType fromObject(Object recordObj, string label = "", IDictionary<string, IList<string>> properties = null)
        {
            // Create the type schema from the object
            // --------------------------------------
            // Get the class type
            Type object_type = recordObj.GetType();

            return fromClass( object_type, label, properties );
        }  // end fromObject()

        public KineticaType(IList<Column> columns)
        {
            m_data.columns = columns;
            Initialize();
            CreateSchema();  // create the schema from columns
        }

        public KineticaType(string label, IList<Column> columns) : this(columns)
        {
            m_data.label = label;
        }

        public KineticaType( string label, IList<Column> columns, IDictionary<string, IList<string>> properties ) : this( label, columns )
        {
            m_properties = properties ?? new Dictionary<string, IList<string>>();
        }

        public KineticaType(string typeSchema)
        {
            m_data.schemaString = typeSchema;
            CreateSchemaFromString( typeSchema );
            CreateSchema();
        }

        public KineticaType(string label, string typeSchema, IDictionary<string, IList<string>> properties, string? typeId = null )
        {
            m_properties = properties;
            m_typeId = typeId;
            m_data.label = label;
            m_data.schemaString = typeSchema;
            CreateSchemaFromString(typeSchema, properties);
            CreateSchema();
        }

        public string getLabel() { return m_data.label; }
        public IList<Column> getColumns() { return m_data.columns; }
        public Column getColumn(int index) { return m_data.columns[index]; }
        public Column getColumn(string name) { return m_data.columns[getColumnIndex(name)]; }
        public int getColumnCount() { return m_data.columns.Count; }
        public int getColumnIndex(string name) { return m_data.columnMap[name]; }
        public bool hasColumn(string name) { return m_data.columnMap.ContainsKey(name); }
        public Schema getSchema() { return m_data.schema; }
        public Type? getSourceType() { return m_data.sourceType;}
        public string getSchemaString() { return m_data.schemaString; }
        public string getTypeID() { return m_typeId;  }

        public void saveSourceType( Type sourceType )
        {
            this.m_data.sourceType = sourceType;
        }  // end saveSourceType


        public string create(Kinetica kinetica)
        {
            // Save the association between this KineticaType's source and itself in the Kinetica object
            // for future reference (it helps with encoding and decoding records)
            if ( this.m_data.sourceType != null )
                kinetica.SetKineticaSourceClassToTypeMapping( this.m_data.sourceType, this );

            // Register the type with Kinetica
            CreateTypeResponse response = kinetica.createType( m_data.schemaString, m_data.label, m_properties);
            return response.type_id;
        }  // end create()

        private KineticaType() { }

        private void Initialize()
        {
            int columnCount = m_data.columns.Count;

            if (columnCount == 0)
            {
                throw new ArgumentException("At least one column must be specified.");
            }

            for (int i = 0; i < columnCount; ++i)
            {
                string columnName = m_data.columns[i].getName();

                if (m_data.columnMap.ContainsKey(columnName))
                {
                    throw new ArgumentException("Duplicate column name " + columnName + " specified.");
                }

                m_data.columnMap[columnName] = i;
            }
        }  // end Initialize()


        private void CreateSchemaFromString( string typeSchema,
                                             IDictionary<string, IList<string>> properties = null)
        {
            // Create the avro schema from the string and save it
            try
            {
                m_data.schema = RecordSchema.Parse(typeSchema);
            }
            catch (Exception ex)
            {
                throw new KineticaException(ex.ToString());
            }

            var root = JObject.Parse(typeSchema);

            var rootType = root["type"];
            if ((null == rootType) || !rootType.ToString().Contains("record"))
            {
                throw new ArgumentException("Schema must be of type record.");
            }

            var fields = root["fields"];
            if ((null == fields) || !fields.HasValues)
            {
                throw new ArgumentException("Schema has no fields.");
            }

            foreach (var field in fields)
            {
                //if (!field->first.empty() || field->second.empty())
                //{
                //    throw std::invalid_argument("Schema has invalid field.");
                //}

                // Do NOT use ToString 'cause it includes the double quotes (turns it into a JSON representation)
                string? fieldName = (string?)field["name"];
                if (string.IsNullOrEmpty(fieldName))
                {
                    throw new ArgumentException("Schema has unnamed field.");
                }

                if (m_data.columnMap.ContainsKey(fieldName))
                {
                    throw new ArgumentException($"Duplicate field name {fieldName}.");
                }

                var fieldType = field["type"];
                if (null == fieldType)
                {
                    throw new ArgumentException($"Field {fieldName} has no type.");
                }

                // Flag for nullability
                bool isColumnNullable = false;

                if (fieldType.HasValues) // If it has children
                {
                    var fieldTypeArray = fieldType;

                    foreach (var fieldTypeElement in fieldTypeArray.Children())
                    {
                        bool valid = false;
                        //if (fieldTypeElement->first.empty())
                        {
                            var fieldTypeElementString = fieldTypeElement.ToString();

                            if (!string.IsNullOrEmpty(fieldTypeElementString))
                            {
                                if (fieldTypeElementString == "null" || fieldTypeElementString == "\"null\"")
                                {
                                    isColumnNullable = true;
                                    valid = true;
                                }
                                else //if (fieldType->empty())
                                {
                                    fieldType = fieldTypeElement; // fieldTypeElementString;
                                    valid = true;
                                }
                            }
                        }

                        if (!valid)
                        {
                            throw new ArgumentException("Field {fieldName} has invalid type.");
                        }
                    }
                }

                Column.ColumnType columnType;

                if (fieldType.ToString().Equals("bytes") || fieldType.ToString().Equals("\"bytes\""))
                {
                    columnType = Column.ColumnType.BYTES;
                }
                else if (fieldType.ToString().Equals("double") || fieldType.ToString().Equals("\"double\""))
                {
                    columnType = Column.ColumnType.DOUBLE;
                }
                else if (fieldType.ToString().Equals("float") || fieldType.ToString().Equals("\"float\""))
                {
                    columnType = Column.ColumnType.FLOAT;
                }
                else if (fieldType.ToString().Equals("int") || fieldType.ToString().Equals("\"int\""))
                {
                    columnType = Column.ColumnType.INT;
                }
                else if (fieldType.ToString().Equals("long") || fieldType.ToString().Equals("\"long\""))
                {
                    columnType = Column.ColumnType.LONG;
                }
                else if (fieldType.ToString().Equals("string") || fieldType.ToString().Equals("\"string\""))
                {
                    columnType = Column.ColumnType.STRING;
                }
                else
                {
                    throw new ArgumentException("Field {fieldName} must be of type bytes, double, float, int, long or string.");
                }

                IList<string>? columnProperties = null;
                properties?.TryGetValue(fieldName, out columnProperties);
                // Check the column properties for nullability
                if ( ( null != columnProperties ) &&
                     ( columnProperties.Contains( ColumnProperty.NULLABLE ) ) )
                    isColumnNullable = true;

                // Create the column to be added
                Column column = new( fieldName, columnType, columnProperties );

                column.setIsNullable( isColumnNullable );

                m_data.columns.Add( column );

                m_data.columnMap[fieldName] = m_data.columns.Count - 1;
            }
        }  // end CreateSchemaFromString()

        private void CreateSchema()
        {
            // First, check if the schema has already been created
            if (m_data.schema != null)
            {
                // nothing to do
                return;
            }

            // Check if the schema string exists, if so, create the schema from that
            if (m_data.schemaString != null)
            {
                try
                {
                    m_data.schema = RecordSchema.Parse(m_data.schemaString);
                    return;
                }
                catch (Exception ex)
                {
                    throw new KineticaException(ex.ToString());
                }
            }  // done creating the schema from the schema string

            // Since the shortcuts didn't apply, create a JSON object from the columns
            // and then create the schema and the schema string off it
            // --------------------------------------------------------------------------
            // Create the json string for the type
            string schemaString = "";
            // Create the json string opening with empty fields (with a generic 'type_name' (because the
            // server always replaces the name with this string anyway) )
            string schemaOpening = "{'type':'record','name':'type_name','fields':[";
            // Create the json string closing
            string schemaClosing = "]}";

            schemaString += schemaOpening;

            // Create the json substrings for the columns
            foreach (var column in m_data.columns)
            {
                // Add the name
                string fieldName = ("'name':'" + column.getName() + "'");

                // Add the type
                string fieldType = "";
                if (column.isNullable())
                {  // the column is nullable, so we need a union
                    fieldType = ("['" + column.getTypeString() + "','null']");
                }
                else  // regular type, no union needed
                {
                    fieldType = ( "'" + column.getTypeString() + "'" );
                }
                fieldType = ("'type':" + fieldType);

                // Put the field together
                string field = ("{" + fieldName + "," + fieldType + "},");
                schemaString += field;
            }  // end looping over the fields

            // Trim the trailing comma from the fields
            char[] comma = [','];
            schemaString = schemaString.TrimEnd(comma);
            // Add the ending of the json string
            schemaString += schemaClosing;

            // Create the RecordSchema from the JSON string
            try
            {
                m_data.schema = RecordSchema.Parse(schemaString);
            }
            catch (Exception ex)
            {
                throw new KineticaException(ex.ToString());
            }

            // Save the schema string
            m_data.schemaString = m_data.schema.ToString();
            return;
        }  // end CreateSchema()
    }  // end class KineticaType
}  // end namespace kinetica