RecordKey.cs

Go to the documentation of this file.

using System;
using System.Collections.Generic;
using System.Numerics;
using System.Text.RegularExpressions;


namespace kinetica.Utils;

    internal sealed class RecordKey
    {
        private static readonly Regex DATE_REGEX = new Regex("\\A(\\d{4})-(\\d{2})-(\\d{2})$");

        private static readonly Regex DATETIME_REGEX = new Regex("\\A(?<year>\\d{4})-(?<month>\\d{2})-(?<day>\\d{2})(?<time>\\s+(?<hour>\\d{1,2}):(?<min>\\d{2}):(?<sec>\\d{2})(?:\\.(?<ms>\\d{1,6}))?)?$");


        private static readonly Regex IPV4_REGEX = new Regex("\\A(?<a>\\d{1,3})\\.(?<b>\\d{1,3})\\.(?<c>\\d{1,3})\\.(?<d>\\d{1,3})$");

        private static readonly Regex TIME_REGEX = new Regex("\\A(?<hour>\\d{1,2}):(?<minute>\\d{2}):(?<seconds>\\d{2})(\\.(?<milliseconds>\\d{1,3}))?$");

        private static readonly DateTime EPOCH_DATE = new DateTime(1970, 1, 1);

        private static readonly int MIN_SUPPORTED_YEAR = 1000;

        private static readonly int MAX_SUPPORTED_YEAR = 2900;

        private static readonly int YEAR_1900 = 1900;

        private static readonly TimeZoneInfo UTC = TimeZoneInfo.Utc;

        private readonly byte[] buffer;
        private readonly int buffer_size;
        private int current_size;
        private int hash_code;
        private bool is_valid;
        private long routingHash;

        public RecordKey(int size)
        {
            if (size < 1)
                throw new KineticaException("Buffer size must be greater than or equal to 1.  "
                                              + "Size given: " + size);
            buffer_size = size;
            current_size = 0;
            buffer = new byte[size];
            this.is_valid = true;
        }

        public bool isValid()
        {
            return this.is_valid;
        }

        public int hashCode()
        {
            return this.hash_code;
        }



        private bool isBufferFull(bool throw_if_full = true)
        {
            if (this.current_size == this.buffer_size)
            {
                if (throw_if_full)
                    throw new KineticaException("The buffer is already full!");
                return true; // yes, the buffer is full, and we haven't thrown
            }
            return false; // buffer is NOT full
        }  // end isBufferFull

        private bool willBufferOverflow(int n, bool throw_if_overflow = true)
        {
            // Note: We're not checking for a negative value for n here
            if ((this.current_size + n) > this.buffer_size)
            {
                if (throw_if_overflow)
                    throw new KineticaException($"The buffer (of size {buffer_size}) does not have sufficient room in it to put {n} more byte(s) (current size is {this.current_size}).");
                return true; // yes, the buffer WILL overflow, but we haven't thrown
            }
            return false; // buffer will NOT overflow
        }  // end willBufferOverflow


        private void add(byte b)
        {
            // Add the byte to the buffer and increment the size
            buffer.SetValue(b, current_size++);
        } // end add()



        public void addInt(int? value)
        {
            // Check if the given number of characters will fit in the buffer
            this.willBufferOverflow(4);  // int is four bytes long

            // Handle nulls
            if (value == null)
            {
                // Add four zero bytes for the null value
                this.add((byte)0);  // 1st 0
                this.add((byte)0);  // 2nd 0
                this.add((byte)0);  // 3rd 0
                this.add((byte)0);  // 4th 0
                return;
            }

            // Put the integer into the array, but first convert to bytes
            byte[] int_bytes = BitConverter.GetBytes((int)value);

            // Add the four bytes
            foreach (byte b in int_bytes)
                this.add(b);
        }  // end addInt


        public void addInt8(int? value)
        {
            // Check if the given number of characters will fit in the buffer
            this.willBufferOverflow(1);  // int8 is one byte long

            // Handle nulls
            if (value == null)
            {
                // Add one zero byte for the null value
                this.add((byte)0);
                return;
            }

            // Put the integer into the array, but first convert to byte
            this.add((byte)value);
        }  // end addInt8


        public void addInt16(int? value)
        {
            // Check if the given number of characters will fit in the buffer
            this.willBufferOverflow(2);  // int16 is two bytes long

            // Handle nulls
            if (value == null)
            {
                // Add two zero bytes for the null value
                this.add((byte)0);  // 1st 0
                this.add((byte)0);  // 2nd 0
                return;
            }

            // Put the short into the array, but first convert to bytes
            byte[] short_bytes = BitConverter.GetBytes((short)value);

            // Add the two bytes
            foreach (byte b in short_bytes)
                this.add(b);
        }  // end addInt16



        public void addLong(long? value)
        {
            // Check if the given number of characters will fit in the buffer
            this.willBufferOverflow(8);  // int is eight bytes long

            // Handle nulls
            if (value == null)
            {
                // Add four zero bytes for the null value
                this.add((byte)0);  // 1st 0
                this.add((byte)0);  // 2nd 0
                this.add((byte)0);  // 3rd 0
                this.add((byte)0);  // 4th 0
                this.add((byte)0);  // 5th 0
                this.add((byte)0);  // 6th 0
                this.add((byte)0);  // 7th 0
                this.add((byte)0);  // 8th 0
                return;
            }

            // Put the long into the array, but first convert to bytes
            byte[] long_bytes = BitConverter.GetBytes((long)value);

            // Add the eight bytes
            foreach (byte b in long_bytes)
                this.add(b);
        }  // end addLong


        public void addFloat(float? value)
        {
            // Check if the given number of characters will fit in the buffer
            this.willBufferOverflow(4);  // int is four bytes long

            // Handle nulls
            if (value == null)
            {
                // Add four zero bytes for the null value
                this.add((byte)0.0f);  // 1st 0
                this.add((byte)0.0f);  // 2nd 0
                this.add((byte)0.0f);  // 3rd 0
                this.add((byte)0.0f);  // 4th 0
                return;
            }

            // Put the integer into the array, but first convert to bytes
            byte[] float_bytes = BitConverter.GetBytes((float)value);

            // Add the four bytes
            foreach (byte b in float_bytes)
                this.add(b);
        }  // end addFloat



        public void addDouble(double? value)
        {
            // Check if the given number of characters will fit in the buffer
            this.willBufferOverflow(8);  // int is eight bytes long

            // Handle nulls
            if (value == null)
            {
                // Add four zero bytes for the null value
                this.add((byte)0.0);  // 1st 0
                this.add((byte)0.0);  // 2nd 0
                this.add((byte)0.0);  // 3rd 0
                this.add((byte)0.0);  // 4th 0
                this.add((byte)0.0);  // 5th 0
                this.add((byte)0.0);  // 6th 0
                this.add((byte)0.0);  // 7th 0
                this.add((byte)0.0);  // 8th 0
                return;
            }

            // Put the integer into the array, but first convert to bytes
            byte[] double_bytes = BitConverter.GetBytes((double)value);

            // Add the eight bytes
            foreach (byte b in double_bytes)
                this.add(b);
        }  // end addDouble



        public void addString(string value)
        {
            // Handle nulls
            if (value == null)
            {
                this.addLong(0L);
                return;
            }

            // Hash the value
            MurMurHash3.LongPair murmur = new MurMurHash3.LongPair();
            System.Text.Encoding encoding = new System.Text.UTF8Encoding();
            byte[] input = encoding.GetBytes(value);
            MurMurHash3.murmurhash3_x64_128(input, 0, (uint)input.Length, 10, out murmur);

            // Add the hashed value to the buffer
            this.addLong(murmur.val1);
        }  // end addString



        public void addCharN(string value, int N)
        {
            // Check if the given number of characters will fit in the buffer
            this.willBufferOverflow(N);
            //if ( ( this.current_size + N ) > buffer_size )
            //    throw new KineticaException( $"The given {N} character(s) will not fit in the buffer (of size {buffer_size}) which has {this.current_size} bytes in it already." );

            // Handle nulls
            if (value == null)
            {
                for (int i = 0; i < N; ++i)
                {
                    this.add((byte)0);
                }
                return;
            }

            // Encode the string into bytes (using the UTF-8 encoding)
            byte[] bytes = System.Text.Encoding.UTF8.GetBytes(value);
            int byte_count = bytes.GetLength(0);

            // Truncate longer strings to the given length
            if (byte_count > N)
                byte_count = N;

            // Put the characters in the byte buffer in the little endian
            // order (which means it will be right to left)
            // ----------------------------------------------------------
            // First, pad with any zeroes "at the end"
            for (int i = N; i > byte_count; --i)
            {
                this.add((byte)0);
            }

            // Then, put all the characters (in reverse order)
            for (int i = (byte_count - 1); i >= 0; --i)
            {
                this.add(bytes[i]);
            }
        }  // end addCharN()


        public void addDate(string value)
        {
            // Check and throw if the buffer is already full
            this.isBufferFull(true);

            // Handle nulls
            if (value == null)
            {
                this.addInt(0);
                return;
            }

            // Check that the given value matches the YYYY-MM-DD pattern
            Match match = DATE_REGEX.Match(value);
            if (!match.Success)
            {
                // No match, so the key is invalid
                this.is_valid = false;
                this.addInt(0);
                return;
            }

            // We'll need to parse the string into year, month, and day
            int year, month, day;
            DateTime date;
            System.Globalization.GregorianCalendar calendar = new System.Globalization.GregorianCalendar();

            // Parse the string value
            try
            {
                year = int.Parse(match.Groups[1].ToString());
                month = int.Parse(match.Groups[2].ToString());
                day = int.Parse(match.Groups[3].ToString());
                date = new DateTime(year, month, day, calendar);
            }
            catch (Exception)
            {
                // Upon any error, set this key to be invalid
                this.addInt(0);
                this.is_valid = false;
                return;
            }

            // Kinetica does not support years outside the range [1000, 2900]
            if ((year < MIN_SUPPORTED_YEAR) || (year > MAX_SUPPORTED_YEAR))
            {
                this.addInt(0);
                this.is_valid = false;
                return;
            }

            int fixed_day_of_week = ((int)calendar.GetDayOfWeek(date) + 1);

            // Deduce the integer representing the date
            int date_integer = (((year - YEAR_1900) << 21)
                                 | (month << 17)
                                 | (day << 12)
                                 | (calendar.GetDayOfYear(date) << 3)
                                 | fixed_day_of_week);
            this.addInt(date_integer);
        }  // end addDate()


        public void addDateTime(string value)
        {
            // Check and throw if the buffer is already full
            this.isBufferFull(true);

            // Handle nulls
            if (value == null)
            {
                this.addLong(0);
                return;
            }

            // Check that the given value matches the YYYY-MM-DD HH:MM:SS.mmm pattern
            Match match = DATETIME_REGEX.Match(value);
            if (!match.Success)
            {
                // No match, so the key is invalid
                this.is_valid = false;
                this.addLong(0);
                return;
            }

            // We'll need to parse the string into year, month, day, hour,
            // minute, second, and millisecond
            int year, month, day;
            int hour = 0;
            int minute = 0;
            int second = 0;
            int msecond = 0;
            DateTime date;
            System.Globalization.GregorianCalendar calendar = new System.Globalization.GregorianCalendar();

            // Parse the string value
            try
            {
                year = int.Parse(match.Groups["year"].Value);
                month = int.Parse(match.Groups["month"].Value);
                day = int.Parse(match.Groups["day"].Value);

                // Handle the optional time part
                Group time_group = match.Groups["time"];
                if (time_group.Success)
                {
                    hour = int.Parse(match.Groups["hour"].Value);
                    minute = int.Parse(match.Groups["min"].Value);
                    second = int.Parse(match.Groups["sec"].Value);

                    // Handle the further optional milliseconds
                    Group ms_group = match.Groups["ms"];
                    if (ms_group.Success)
                    {
                        msecond = int.Parse(match.Groups["ms"].Value);
                        // Need to have the milliseconds be milliseconds (three digits)
                        switch (ms_group.Value.Length)
                        {
                            case 1:
                                msecond *= 100; break;
                            case 2:
                                msecond *= 10; break;
                            // No need for case 3
                            case 4:
                                msecond /= 10; break;
                            case 5:
                                msecond /= 100; break;
                            case 6:
                                msecond /= 1000; break;
                        }
                    }
                }  // end parsing the time component

                // Now put it all together
                date = new DateTime(year, month, day, hour, minute, second, msecond, calendar);
            }
            catch (Exception)
            {
                // Upon any error, set this key to be invalid
                this.addLong(0);
                this.is_valid = false;
                return;
            }

            // Kinetica does not support years outside the range [1000, 2900]
            if ((year < MIN_SUPPORTED_YEAR) || (year > MAX_SUPPORTED_YEAR))
            {
                this.addLong(0);
                this.is_valid = false;
                return;
            }

            int fixed_day_of_week = ((int)calendar.GetDayOfWeek(date) + 1);

            // Deduce the integer representing the date
            long datetime_long = (long)((((long)(year - YEAR_1900)) << 53)
                                          | (((long)month) << 49)
                                          | (((long)day) << 44)
                                          | (((long)hour) << 39)
                                          | (((long)minute) << 33)
                                          | (((long)second) << 27)
                                          | (((long)msecond) << 17)
                                          | (((long)calendar.GetDayOfYear(date)) << 8)
                                          | (((long)fixed_day_of_week) << 5));
            this.addLong(datetime_long);
        }  // end addDateTime()


        public void addDecimal(string value, int precision, int scale)
        {
            // Check and throw if the buffer is already full
            this.isBufferFull(true);

            // Determine byte size based on precision
            int byteSize = (precision > 18) ? 12 : 8;

            // Handle nulls or empty strings
            if (string.IsNullOrEmpty(value) || value == "null")
            {
                for (int i = 0; i < byteSize; i++)
                    this.add((byte)0);
                return;
            }

            try
            {
                // Parse the decimal value
                if (!decimal.TryParse(value, System.Globalization.NumberStyles.Any,
                    System.Globalization.CultureInfo.InvariantCulture, out decimal parsedValue))
                {
                    // Invalid format - write zeros and mark invalid
                    for (int i = 0; i < byteSize; i++)
                        this.add((byte)0);
                    this.is_valid = false;
                    return;
                }

                // Calculate the scale multiplier
                decimal multiplier = (decimal)Math.Pow(10, scale);

                // Scale the value and round to get unscaled integer
                decimal scaledValue = Math.Round(parsedValue * multiplier);

                if (precision <= 18)
                {
                    // 8-byte decimal: store as long
                    long longValue = (long)scaledValue;
                    this.addLong(longValue);
                }
                else
                {
                    // 12-byte decimal: store as BigInteger in little-endian
                    this.willBufferOverflow(12);

                    // Convert to BigInteger
                    BigInteger bigValue = new BigInteger(scaledValue);

                    // Get little-endian bytes, pad or truncate to 12 bytes
                    byte[] bigBytes = bigValue.ToByteArray();
                    byte[] result = new byte[12];

                    // Copy bytes, handling both positive and negative numbers
                    int copyLen = Math.Min(bigBytes.Length, 12);
                    Array.Copy(bigBytes, result, copyLen);

                    // Sign-extend if negative and bytes are fewer than 12
                    if (bigValue < 0 && bigBytes.Length < 12)
                    {
                        for (int i = bigBytes.Length; i < 12; i++)
                            result[i] = 0xFF;
                    }

                    // Add all 12 bytes
                    foreach (byte b in result)
                        this.add(b);
                }
            }
            catch (Exception)
            {
                // Upon any error, set this key to be invalid
                for (int i = 0; i < byteSize; i++)
                    this.add((byte)0);
                this.is_valid = false;
            }
        }  // end addDecimal(value, precision, scale)


        public void addIPv4(string value)
        {
            // Check and throw if the buffer is already full
            this.isBufferFull(true);

            // Handle nulls
            if (value == null)
            {
                this.addInt(0);
                return;
            }

            // Check that the given value matches the XXX.XXX.XXX.XXX pattern
            Match match = IPV4_REGEX.Match(value);
            if (!match.Success)
            {
                // No match, so the key is invalid
                this.is_valid = false;
                this.addInt(0);
                return;
            }

            // We'll need to parse the string into four integers
            int a, b, c, d;

            // Parse the string value
            try
            {
                a = int.Parse(match.Groups["a"].Value);
                b = int.Parse(match.Groups["b"].Value);
                c = int.Parse(match.Groups["c"].Value);
                d = int.Parse(match.Groups["d"].Value);
            }
            catch (Exception)
            {
                // Upon any error, set this key to be invalid
                this.addInt(0);
                this.is_valid = false;
                return;
            }

            // Each byte has to be within the range [0, 255] (the regex does
            // not support negative numbers, so no worries about those)
            if ((a > 255) || (b > 255) || (c > 255) || (d > 255))
            {
                this.addInt(0);
                this.is_valid = false;
                return;
            }

            // Deduce the integer representing the date
            int ipv4_integer = ((a << 24) | (b << 16) | (c << 8) | d);
            this.addInt(ipv4_integer);
        }  // end addIPv4()


        public void addTime(string value)
        {
            // Check and throw if the buffer is already full
            this.isBufferFull(true);

            // Handle nulls
            if (value == null)
            {
                this.addInt(0);
                return;
            }

            // Check that the given value matches the HH:MM:SS[.mmm] pattern
            Match match = TIME_REGEX.Match(value);
            if (!match.Success)
            {
                // No match, so the key is invalid
                this.is_valid = false;
                this.addInt(0);
                return;
            }

            // We'll need to parse the string into four integers
            uint hour, minute, second, milliseconds;

            // Parse the string value
            try
            {
                hour = uint.Parse(match.Groups["hour"].Value);
                minute = uint.Parse(match.Groups["minute"].Value);
                second = uint.Parse(match.Groups["seconds"].Value);
                Group msec_group = match.Groups["milliseconds"];

                // Milliseconds are optional
                milliseconds = 0;
                if (msec_group.Success)
                {
                    milliseconds = uint.Parse(msec_group.Value);

                    // Handle single and double digits for milliseconds
                    switch (msec_group.Value.Length)
                    {
                        case 1:
                            milliseconds *= 100; break;
                        case 2:
                            milliseconds *= 10; break;
                    }
                }
            }
            catch (Exception)
            {
                // Upon any error, set this key to be invalid
                this.addInt(0);
                this.is_valid = false;
                return;
            }

            // Validate the hour, minute, second values
            if ((hour > 23) || (minute > 59) || (second > 59))
            {
                this.addInt(0);
                this.is_valid = false;
                return;
            }

            // Deduce the integer representing the time
            int time_integer = (int)((hour << 26) | (minute << 20) | (second << 14) | (milliseconds << 4));
            this.addInt(time_integer);
        }  // end addTime()


        public void addTimeStamp(long? value)
        {
            // Handle nulls
            if (value == null)
            {
                this.addLong(0);
                return;
            }

            // Encode the timestamp the way the database server does it
            DateTime time = EPOCH_DATE.AddMilliseconds((double)value);
            long fixed_day_of_week = ((long)time.DayOfWeek + 1);

            long timestamp = (long)((((long)(time.Year - YEAR_1900)) << 53)
                                      | (((long)(time.Month)) << 49)
                                      | (((long)time.Day) << 44)
                                      | (((long)time.Hour) << 39)
                                      | (((long)time.Minute) << 33)
                                      | (((long)time.Second) << 27)
                                      | (((long)time.Millisecond) << 17)
                                      | (((long)time.DayOfYear) << 8)
                                      | (fixed_day_of_week << 5));
            this.addLong(timestamp);
        }  // end addTimeStamp()



        public void computeHashes()
        {
            // Check all the values for the key have been added
            if (this.current_size != this.buffer_size)
                throw new KineticaException("The RecordKey buffer is not full; check that all the relevant values have been added.");

            // Hash the value
            MurMurHash3.LongPair murmur = new MurMurHash3.LongPair();
            MurMurHash3.murmurhash3_x64_128(this.buffer, 0, (uint)this.buffer_size, 10, out murmur);

            // Save the hash value
            this.routingHash = murmur.val1;
            this.hash_code = (int)(this.routingHash ^ ((this.routingHash >> 32) & 0x0000ffffL));
        }  // end computeHashes



        public int route(IList<int> routingTable)
        {
            // Return 1 less than the value of the nth element of routingTable where
            //    n == (record key hash) % (number of elements in routingTable)
            // (because the 1st worker rank is the 0th element in the worker list)
            return (routingTable[Math.Abs((int)(this.routingHash % routingTable.Count))] - 1);
        }  // end route

    }  // end class RecordKey