ProtocolWireFormatTests.cs

Go to the documentation of this file.

using System;
using System.Linq;
using System.Text;
using Xunit;

namespace Kinetica.Tests.UnitTests
{
    [Trait("Category", "WireFormat")]
    [Trait("Category", "Protocol")]
    public class ProtocolWireFormatTests
    {
        #region RecordKey Wire Format Tests

        [Fact]
        public void RecordKey_IntegerShardKey_ProducesStableEncoding()
        {
            // Arrange: Create a RecordKey with a specific integer value
            var recordKeyType = typeof(kinetica.Kinetica).Assembly
                .GetType("kinetica.Utils.RecordKey");
            var recordKey = Activator.CreateInstance(recordKeyType, 4); // 4-byte buffer

            var addIntMethod = recordKeyType.GetMethod("addInt");
            addIntMethod!.Invoke(recordKey, new object[] { 12345 });

            // Act: Get the binary representation
            var bufferField = recordKeyType.GetField("buffer",
                System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance);
            var buffer = (byte[])bufferField!.GetValue(recordKey)!;

            // Assert: Integer 12345 should be encoded as little-endian
            // 12345 = 0x00003039 in little-endian: 0x39, 0x30, 0x00, 0x00
            byte[] expected = { 0x39, 0x30, 0x00, 0x00 };
            Assert.Equal(expected, buffer);
        }

        [Fact]
        public void RecordKey_StringShardKey_ProducesStableEncoding()
        {
            // Arrange
            var recordKeyType = typeof(kinetica.Kinetica).Assembly
                .GetType("kinetica.Utils.RecordKey");
            var recordKey = Activator.CreateInstance(recordKeyType, 8); // 8-byte buffer for char8

            var addCharNMethod = recordKeyType.GetMethod("addCharN");
            addCharNMethod!.Invoke(recordKey, new object[] { "ABCD", 8 });

            // Act
            var bufferField = recordKeyType.GetField("buffer",
                System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance);
            var buffer = (byte[])bufferField!.GetValue(recordKey)!;

            // Assert: "ABCD" as UTF-8 with little-endian encoding (null padding first, then bytes in reverse)
            byte[] expected = { 0x00, 0x00, 0x00, 0x00, 0x44, 0x43, 0x42, 0x41 };  // [null padding, D, C, B, A]
            Assert.Equal(expected, buffer);
        }

        [Fact]
        public void RecordKey_LongShardKey_ProducesStableEncoding()
        {
            // Arrange
            var recordKeyType = typeof(kinetica.Kinetica).Assembly
                .GetType("kinetica.Utils.RecordKey");
            var recordKey = Activator.CreateInstance(recordKeyType, 8);

            var addLongMethod = recordKeyType.GetMethod("addLong");
            addLongMethod!.Invoke(recordKey, new object[] { 9876543210L });

            // Act
            var bufferField = recordKeyType.GetField("buffer",
                System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance);
            var buffer = (byte[])bufferField!.GetValue(recordKey)!;

            // Assert: 9876543210 = 0x0000000024CB016EA, BitConverter encodes as little-endian
            byte[] expected = { 234, 22, 176, 76, 2, 0, 0, 0 };  // [0xEA, 0x16, 0xB0, 0x4C, 0x02, 0x00, 0x00, 0x00]
            Assert.Equal(expected, buffer);
        }

        [Fact]
        public void RecordKey_Decimal8Byte_StableWireFormat()
        {
            // Arrange: Decimal value "123.4567" with scale 4
            var recordKeyType = typeof(kinetica.Kinetica).Assembly
                .GetType("kinetica.Utils.RecordKey");
            var recordKey = Activator.CreateInstance(recordKeyType, 8);

            var addDecimalMethod = recordKeyType.GetMethod("addDecimal",
                new[] { typeof(string), typeof(int), typeof(int) });
            addDecimalMethod!.Invoke(recordKey, new object[] { "123.4567", 18, 4 });

            // Act
            var bufferField = recordKeyType.GetField("buffer",
                System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance);
            var buffer = (byte[])bufferField!.GetValue(recordKey)!;

            // Assert: 123.4567 with scale 4 = 1234567 as long
            // 1234567 = 0x0012D687 in little-endian
            long expected = 1234567L;
            long actual = BitConverter.ToInt64(buffer, 0);
            Assert.Equal(expected, actual);
        }

        [Fact]
        public void RecordKey_Decimal12Byte_StableWireFormat()
        {
            // Arrange: Large decimal requiring 12-byte encoding
            var recordKeyType = typeof(kinetica.Kinetica).Assembly
                .GetType("kinetica.Utils.RecordKey");
            var recordKey = Activator.CreateInstance(recordKeyType, 12);

            var addDecimalMethod = recordKeyType.GetMethod("addDecimal",
                new[] { typeof(string), typeof(int), typeof(int) });
            // Use a value that requires more than 8 bytes
            addDecimalMethod!.Invoke(recordKey, new object[] { "12345678901234567890.123456", 38, 6 });

            // Act
            var bufferField = recordKeyType.GetField("buffer",
                System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance);
            var buffer = (byte[])bufferField!.GetValue(recordKey)!;
            var currentSizeField = recordKeyType.GetField("current_size",
                System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance);
            var size = (int)currentSizeField!.GetValue(recordKey)!;

            // Assert: Should use all 12 bytes
            Assert.Equal(12, size);

            // Verify buffer is not all zeros (contains the large value)
            bool hasNonZero = buffer.Any(b => b != 0);
            Assert.True(hasNonZero, "12-byte decimal buffer should contain non-zero bytes");
        }

        #endregion

        #region Type Schema Encoding Tests

        [Fact]
        public void KineticaType_SimpleSchema_ProducesStableSchemaString()
        {
            // Arrange: Define a simple type
            string schemaJson = @"{
                ""type"": ""record"",
                ""name"": ""TestRecord"",
                ""fields"": [
                    {""name"": ""id"", ""type"": ""int""},
                    {""name"": ""name"", ""type"": ""string""}
                ]
            }";

            // Act: Create type from schema
            var ktype = new kinetica.KineticaType(schemaJson);
            string generatedSchema = ktype.getSchemaString();

            // Assert: Schema should match expected format (normalized JSON)
            // The exact format may vary, but key elements must be present
            Assert.Contains("\"type\":\"record\"", generatedSchema.Replace(" ", ""));
            Assert.Contains("\"name\":\"TestRecord\"", generatedSchema.Replace(" ", ""));
            Assert.Contains("\"fields\"", generatedSchema);
            Assert.Contains("\"id\"", generatedSchema);
            Assert.Contains("\"name\"", generatedSchema);
        }

        [Fact]
        public void KineticaType_NullableField_UsesConsistentUnionSchema()
        {
            // Arrange
            string schemaJson = @"{
                ""type"": ""record"",
                ""name"": ""NullableTest"",
                ""fields"": [
                    {""name"": ""optional_int"", ""type"": [""null"", ""int""]}
                ]
            }";

            // Act
            var ktype = new kinetica.KineticaType(schemaJson);
            string generatedSchema = ktype.getSchemaString();

            // Assert: Should contain union type [null, int]
            // Note: Order matters in Avro unions - null first is standard
            Assert.Contains("[\"null\",\"int\"]", generatedSchema.Replace(" ", ""));
        }

        #endregion

        #region Column Property Encoding Tests

        [Fact]
        public void ColumnProperties_ShardKeyProperty_EncodesConsistently()
        {
            // This is a smoke test - actual property encoding happens in server protocol
            // We verify that the constant values don't change
            Assert.Equal("shard_key", kinetica.ColumnProperty.SHARD_KEY);
            Assert.Equal("primary_key", kinetica.ColumnProperty.PRIMARY_KEY);
            Assert.Equal("nullable", kinetica.ColumnProperty.NULLABLE);
            Assert.Equal("dict", kinetica.ColumnProperty.DICT);
            Assert.Equal("timestamp", kinetica.ColumnProperty.TIMESTAMP);
        }

        #endregion

        #region Compression Tests

        [Fact]
        public void Compression_Snappy_RoundTripsCorrectly()
        {
            // Arrange: Sample data
            string originalText = "Hello, Kinetica! This is a test of compression. " +
                                 "The data should compress well due to repetition. " +
                                 "The data should compress well due to repetition.";
            byte[] originalBytes = Encoding.UTF8.GetBytes(originalText);

            // Act: Compress and decompress using Snappier
            byte[] compressed = Snappier.Snappy.CompressToArray(originalBytes);
            byte[] decompressed = Snappier.Snappy.DecompressToArray(compressed);

            // Assert: Round-trip should preserve data exactly
            Assert.Equal(originalBytes, decompressed);

            // Also verify compression actually reduced size
            Assert.True(compressed.Length < originalBytes.Length,
                $"Compressed size ({compressed.Length}) should be less than original ({originalBytes.Length})");
        }

        #endregion

        #region Regression Tests

        [Fact(Skip = "KineticaType does not support empty schemas - requires at least one field")]
        public void EmptyRecord_EncodesConsistently()
        {
            // Arrange
            string schemaJson = @"{
                ""type"": ""record"",
                ""name"": ""EmptyRecord"",
                ""fields"": []
            }";

            var ktype = new kinetica.KineticaType(schemaJson);

            // Act: Encode an empty record class
            // This tests the edge case of records with no fields
            var schema = ktype.getSchema();

            // Assert: Schema should have zero fields
            var recordSchema = schema as Avro.RecordSchema;
            Assert.NotNull(recordSchema);
            Assert.Empty(recordSchema!.Fields);
        }

        [Fact]
        public void RecordKey_CompositeKey_MaintainsFieldOrder()
        {
            // Arrange: Composite key with int + string
            var recordKeyType = typeof(kinetica.Kinetica).Assembly
                .GetType("kinetica.Utils.RecordKey");
            var recordKey = Activator.CreateInstance(recordKeyType, 12); // 4 bytes int + 8 bytes char8

            var addIntMethod = recordKeyType.GetMethod("addInt");
            var addCharNMethod = recordKeyType.GetMethod("addCharN");

            // Act: Add fields in order
            addIntMethod!.Invoke(recordKey, new object[] { 999 });
            addCharNMethod!.Invoke(recordKey, new object[] { "KEY", 8 });

            // Get buffer
            var bufferField = recordKeyType.GetField("buffer",
                System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance);
            var buffer = (byte[])bufferField!.GetValue(recordKey)!;

            // Assert: First 4 bytes are the int, next 8 bytes are the string
            // 999 = 0x000003E7 in little-endian: 0xE7, 0x03, 0x00, 0x00
            Assert.Equal(0xE7, buffer[0]);
            Assert.Equal(0x03, buffer[1]);
            Assert.Equal(0x00, buffer[2]);
            Assert.Equal(0x00, buffer[3]);

            // "KEY" as UTF-8 with little-endian encoding (padding first: 5 nulls, then Y, E, K in reverse)
            Assert.Equal(0x00, buffer[4]);  // null padding
            Assert.Equal(0x00, buffer[5]);  // null padding
            Assert.Equal(0x00, buffer[6]);  // null padding
            Assert.Equal(0x00, buffer[7]);  // null padding
            Assert.Equal(0x00, buffer[8]);  // null padding
            Assert.Equal((byte)'Y', buffer[9]);   // Reversed
            Assert.Equal((byte)'E', buffer[10]);  // Reversed
            Assert.Equal((byte)'K', buffer[11]);  // Reversed
        }

        #endregion
    }
}