Type.cs

Go to the documentation of this file.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;

namespace kinetica.Records;

    public sealed class Type
    {
        private readonly string _label;
        private readonly IReadOnlyList<Column> _columns;
        private readonly IReadOnlyDictionary<string, int> _columnIndices;
        private readonly IReadOnlyList<int> _shardKeyIndices;
        private readonly IReadOnlyList<int> _primaryKeyIndices;
        private readonly string _schemaString;
        private readonly string? _typeId;

        public Type(string label, IEnumerable<Column> columns)
        {
            _label = label ?? throw new ArgumentNullException(nameof(label));
            var columnList = columns?.ToList() ?? throw new ArgumentNullException(nameof(columns));

            if (columnList.Count == 0)
                throw new ArgumentException("At least one column is required", nameof(columns));

            _columns = columnList.AsReadOnly();

            // Build column index map
            var columnIndices = new Dictionary<string, int>(StringComparer.OrdinalIgnoreCase);
            var shardKeyIndices = new List<int>();
            var primaryKeyIndices = new List<int>();

            for (int i = 0; i < columnList.Count; i++)
            {
                var col = columnList[i];
                if (columnIndices.ContainsKey(col.Name))
                    throw new ArgumentException($"Duplicate column name: {col.Name}", nameof(columns));

                columnIndices[col.Name] = i;

                if (col.IsShardKey)
                    shardKeyIndices.Add(i);

                if (col.IsPrimaryKey)
                    primaryKeyIndices.Add(i);
            }

            _columnIndices = columnIndices;
            _shardKeyIndices = shardKeyIndices.AsReadOnly();
            _primaryKeyIndices = primaryKeyIndices.AsReadOnly();

            // Build Avro schema
            _schemaString = BuildAvroSchema(label, columnList);
            _typeId = null;
        }

        // Private constructor for FromTableResponse
        private Type(
            string label,
            IReadOnlyList<Column> columns,
            IReadOnlyDictionary<string, int> columnIndices,
            IReadOnlyList<int> shardKeyIndices,
            IReadOnlyList<int> primaryKeyIndices,
            string schemaString,
            string? typeId)
        {
            _label = label;
            _columns = columns;
            _columnIndices = columnIndices;
            _shardKeyIndices = shardKeyIndices;
            _primaryKeyIndices = primaryKeyIndices;
            _schemaString = schemaString;
            _typeId = typeId;
        }

        #region Factory Methods

        public static TypeBuilder Builder(string label) => new TypeBuilder(label);

        public static async Task<Type> FromTableAsync(kinetica.Kinetica kinetica, string tableName)
        {
            if (kinetica == null)
                throw new ArgumentNullException(nameof(kinetica));
            if (string.IsNullOrEmpty(tableName))
                throw new ArgumentException("Table name is required", nameof(tableName));

            var response = await Task.Run(() => kinetica.showTable(tableName, new Dictionary<string, string>
            {
                ["get_sizes"] = "false"
            }));

            return FromShowTableResponse(tableName, response);
        }

        public static Type FromTable(kinetica.Kinetica kinetica, string tableName)
        {
            if (kinetica == null)
                throw new ArgumentNullException(nameof(kinetica));
            if (string.IsNullOrEmpty(tableName))
                throw new ArgumentException("Table name is required", nameof(tableName));

            var response = kinetica.showTable(tableName, new Dictionary<string, string>
            {
                ["get_sizes"] = "false"
            });

            return FromShowTableResponse(tableName, response);
        }

        public static Type FromShowTableResponse(string tableName, kinetica.ShowTableResponse response)
        {
            if (response.type_schemas == null || response.type_schemas.Count == 0)
                throw new InvalidOperationException($"No type schema for table '{tableName}'");

            var schemaStr = response.type_schemas[0];

            // Parse the Avro schema JSON
            using var doc = JsonDocument.Parse(schemaStr);
            var root = doc.RootElement;

            var label = root.TryGetProperty("name", out var nameProp)
                ? nameProp.GetString() ?? tableName
                : tableName;

            if (!root.TryGetProperty("fields", out var fieldsProp) || fieldsProp.ValueKind != JsonValueKind.Array)
                throw new InvalidOperationException("Schema missing fields array");

            // Get properties
            if (response.properties == null || response.properties.Count == 0)
                throw new InvalidOperationException("No properties in response");

            var properties = response.properties[0];

            // Build columns
            var columns = new List<Column>();
            var columnIndices = new Dictionary<string, int>(StringComparer.OrdinalIgnoreCase);
            var shardKeyIndices = new List<int>();
            var primaryKeyIndices = new List<int>();

            int idx = 0;
            foreach (var field in fieldsProp.EnumerateArray())
            {
                if (!field.TryGetProperty("name", out var fieldNameProp))
                    throw new InvalidOperationException("Field missing name");

                var fieldName = fieldNameProp.GetString()
                    ?? throw new InvalidOperationException("Field name is null");

                var avroType = ExtractAvroType(field);
                var colProps = properties.TryGetValue(fieldName, out var propList)
                    ? propList
                    : new List<string>();

                var columnType = ColumnTypeExtensions.FromAvroAndProperties(avroType, colProps);
                var column = Column.WithProperties(fieldName, columnType, colProps);

                columns.Add(column);
                columnIndices[fieldName] = idx;

                if (column.IsShardKey)
                    shardKeyIndices.Add(idx);
                if (column.IsPrimaryKey)
                    primaryKeyIndices.Add(idx);

                idx++;
            }

            // Get type ID if available
            string? typeId = null;
            if (response.type_ids != null && response.type_ids.Count > 0)
                typeId = response.type_ids[0];

            return new Type(
                label,
                columns.AsReadOnly(),
                columnIndices,
                shardKeyIndices.AsReadOnly(),
                primaryKeyIndices.AsReadOnly(),
                schemaStr, // Use original schema string from Kinetica
                typeId);
        }

        private static string ExtractAvroType(JsonElement field)
        {
            if (!field.TryGetProperty("type", out var typeProp))
                return "string";

            return typeProp.ValueKind switch
            {
                JsonValueKind.String => typeProp.GetString() ?? "string",
                JsonValueKind.Array => ExtractTypeFromUnion(typeProp),
                JsonValueKind.Object => typeProp.TryGetProperty("type", out var innerType)
                    ? innerType.GetString() ?? "string"
                    : "string",
                _ => "string"
            };
        }

        private static string ExtractTypeFromUnion(JsonElement unionArray)
        {
            // Union type like ["null", "string"] - return the non-null type
            foreach (var element in unionArray.EnumerateArray())
            {
                if (element.ValueKind == JsonValueKind.String)
                {
                    var typeName = element.GetString();
                    if (typeName != null && typeName != "null")
                        return typeName;
                }
            }
            return "string";
        }

        #endregion

        #region Properties

        public string Label => _label;

        public string SchemaString => _schemaString;

        public string? TypeId => _typeId;

        public int ColumnCount => _columns.Count;

        public IReadOnlyList<int> ShardKeyIndices => _shardKeyIndices;

        public IReadOnlyList<int> PrimaryKeyIndices => _primaryKeyIndices;

        public bool HasShardKey => _shardKeyIndices.Count > 0;

        public bool HasPrimaryKey => _primaryKeyIndices.Count > 0;

        public IReadOnlyList<Column> Columns => _columns;

        #endregion

        #region Column Access

        public Column? GetColumn(int index)
        {
            if (index < 0 || index >= _columns.Count)
                return null;
            return _columns[index];
        }

        public Column? GetColumnByName(string name)
        {
            if (_columnIndices.TryGetValue(name, out var index))
                return _columns[index];
            return null;
        }

        public int? GetColumnIndex(string name)
        {
            if (_columnIndices.TryGetValue(name, out var index))
                return index;
            return null;
        }

        public int GetColumnIndexOrThrow(string name)
        {
            if (_columnIndices.TryGetValue(name, out var index))
                return index;
            throw new KeyNotFoundException($"Column not found: {name}");
        }

        #endregion

        #region Record Creation

        public GenericRecord NewInstance() => new GenericRecord(this);

        #endregion

        #region Schema Building

        private static string BuildAvroSchema(string name, IReadOnlyList<Column> columns)
        {
            var sb = new StringBuilder();
            sb.Append("{\"type\":\"record\",\"name\":\"");
            sb.Append(EscapeJsonString(name));
            sb.Append("\",\"fields\":[");

            for (int i = 0; i < columns.Count; i++)
            {
                if (i > 0) sb.Append(',');

                var col = columns[i];
                var avroType = col.AvroTypeName;

                sb.Append("{\"name\":\"");
                sb.Append(EscapeJsonString(col.Name));
                sb.Append("\",\"type\":");

                if (col.IsNullable)
                {
                    sb.Append("[\"null\",\"");
                    sb.Append(avroType);
                    sb.Append("\"]");
                }
                else
                {
                    sb.Append('"');
                    sb.Append(avroType);
                    sb.Append('"');
                }

                sb.Append('}');
            }

            sb.Append("]}");
            return sb.ToString();
        }

        private static string EscapeJsonString(string s)
        {
            // Simple JSON escaping for common cases
            if (s.IndexOfAny(new[] { '"', '\\', '\n', '\r', '\t' }) < 0)
                return s;

            var sb = new StringBuilder(s.Length + 10);
            foreach (var c in s)
            {
                switch (c)
                {
                    case '"': sb.Append("\\\""); break;
                    case '\\': sb.Append("\\\\"); break;
                    case '\n': sb.Append("\\n"); break;
                    case '\r': sb.Append("\\r"); break;
                    case '\t': sb.Append("\\t"); break;
                    default: sb.Append(c); break;
                }
            }
            return sb.ToString();
        }

        #endregion

        public override string ToString()
        {
            return $"Type({_label}, {_columns.Count} columns)";
        }
    }