Fix timestamp size to include alignment, also fix timestamp in union (#…

…656)

src/FlowtideDotNet.Core/ColumnStore/DataColumns/UnionColumn.cs

@@ -14,6 +14,7 @@
 using Apache.Arrow.Memory;
 using Apache.Arrow.Types;
 using FlowtideDotNet.Core.ColumnStore.DataValues;
+using FlowtideDotNet.Core.ColumnStore.Serialization;
 using FlowtideDotNet.Core.ColumnStore.TreeStorage;
 using FlowtideDotNet.Core.ColumnStore.Utils;
 using FlowtideDotNet.Storage.DataStructures;
@@ -335,8 +336,9 @@ public void RemoveAt(in int index)
                 if (_valueColumns[i] != null)
                 {
                     var (arrowArray, arrowType) = _valueColumns[i].ToArrowArray(nullBuffer, nullCount);
+                    var customMetadata = EventArrowSerializer.GetCustomMetadata(arrowType);
                     typeIds.Add((int)arrowType.TypeId);
-                    fields.Add(new Field("", arrowType, true));
+                    fields.Add(new Field("", arrowType, true, metadata: customMetadata));
                     childArrays.Add(arrowArray);
                 }
             }

src/FlowtideDotNet.Core/ColumnStore/DataValues/TimestampTzValue.cs

@@ -43,9 +43,9 @@ public struct TimestampTzValue : IDataValue, IComparable<TimestampTzValue>
         const long UnixEpochTicks = DaysTo1970 * TicksPerDay;
 
         public long ticks;
-        public short offset;
+        public long offset;
 
-        public TimestampTzValue(long ticks, short offset)
+        public TimestampTzValue(long ticks, long offset)
         {
             this.ticks = ticks;
             this.offset = offset;

src/FlowtideDotNet.Core/ColumnStore/Serialization/CustomTypes/TimestampTzType.cs

@@ -25,7 +25,7 @@ internal class TimestampTzType : FixedSizeBinaryType, ICustomArrowType
         public const string ExtensionName = "flowtide.timestamptz";
         public static readonly TimestampTzType Default = new TimestampTzType();
 
-        public TimestampTzType() : base(10)
+        public TimestampTzType() : base(16)
         {
         }
 

src/FlowtideDotNet.Core/ColumnStore/TreeStorage/ArrowToInternalVisitor.cs

@@ -339,11 +339,11 @@ public void Visit(FixedSizeBinaryArray array)
                         _typeId = ArrowTypeId.Timestamp;
                         break;
                     default:
-                        throw new NotImplementedException();
+                        throw new NotImplementedException(typeName);
                 }
                 return;
             }
-            throw new NotImplementedException();
+            throw new NotImplementedException("No metadata field");
         }
     }
 }

tests/FlowtideDotNet.Core.Tests/ColumnStore/ArrowTests.cs

@@ -712,5 +712,31 @@ public void TestDecimalInMapSerializeDeserialize()
             Assert.Equal(2.0m, map);
 
         }
+
+        [Fact]
+        public void TimestampInUnionSerializeDeserialize()
+        {
+            Column column = new Column(GlobalMemoryManager.Instance);
+            column.Add(new TimestampTzValue(1, 0));
+            column.Add(NullValue.Instance);
+            column.Add(new Int64Value(2));
+
+            var recordBatch = EventArrowSerializer.BatchToArrow(new EventBatchData(
+            [
+                column
+            ]), column.Count);
+
+            MemoryStream memoryStream = new MemoryStream();
+
+            var writer = new ArrowStreamWriter(memoryStream, recordBatch.Schema, true);
+            writer.WriteRecordBatch(recordBatch);
+            writer.Dispose();
+            memoryStream.Position = 0;
+            var reader = new ArrowStreamReader(memoryStream, new Apache.Arrow.Memory.NativeMemoryAllocator(), true);
+            var deserializedRecordBatch = reader.ReadNextRecordBatch();
+            var deserializedBatch = EventArrowSerializer.ArrowToBatch(deserializedRecordBatch, GlobalMemoryManager.Instance);
+
+            Assert.Equal(1, column.GetValueAt(0, default).AsTimestamp.ticks);
+        }
     }
 }