[Issue #1358] short-decimal byte ingest and ColumnValue.is_null for sink

pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/DecimalColumnVector.java

@@ -26,6 +26,7 @@
 
 import java.math.BigDecimal;
 import java.math.MathContext;
+import java.nio.ByteBuffer;
 import java.util.Arrays;
 
 import static com.google.common.base.Preconditions.checkArgument;
@@ -152,6 +153,28 @@ public void flatten(boolean selectedInUse, int[] sel, int size)
         flattenNoNulls(selectedInUse, sel, size);
     }
 
+    @Override
+    public void add(byte[] bytes)
+    {
+        if (checkBytesNull(bytes))
+        {
+            return;
+        }
+        if (bytes.length != Long.BYTES)
+        {
+            throw new IllegalArgumentException(
+                    "Short decimal requires exactly " + Long.BYTES + " bytes, got: " + bytes.length);
+        }
+        if (writeIndex >= getLength())
+        {
+            ensureSize(writeIndex * 2, true);
+        }
+        int index = writeIndex++;
+        // The bytes are the big-endian unscaled value of the decimal at this column's scale.
+        vector[index] = ByteBuffer.wrap(bytes).getLong();
+        isNull[index] = false;
+    }
+
     @Override
     public void add(String value)
     {

pixels-core/src/test/java/io/pixelsdb/pixels/core/vector/TestDecimalColumnVector.java

@@ -19,14 +19,32 @@
  */
 package io.pixelsdb.pixels.core.vector;
 
+import io.pixelsdb.pixels.core.TypeDescription;
 import org.junit.Test;
 
+import static org.junit.Assert.assertArrayEquals;
+
 /**
  * Created at: 06/03/2022
  * Author: hank
  */
 public class TestDecimalColumnVector
 {
+    @Test
+    public void testAddCanonicalBytes()
+    {
+        String[] values = {"5755.94", "-283.84", "0", "-2"};
+        TypeDescription type = TypeDescription.createDecimal(15, 2);
+        DecimalColumnVector fromBytes = new DecimalColumnVector(values.length, 15, 2);
+        DecimalColumnVector fromStrings = new DecimalColumnVector(values.length, 15, 2);
+        for (String value : values)
+        {
+            fromBytes.add(type.convertSqlStringToByte(value));
+            fromStrings.add(value);
+        }
+        assertArrayEquals(fromStrings.vector, fromBytes.vector);
+    }
+
     @Test
     public void testAddStringVal()
     {

proto/sink.proto

@@ -55,11 +55,12 @@ message TransactionMetadata {
   string id = 2;
   int64 event_count = 3;
   repeated DataCollection data_collections = 4;  
-  int64 timestamp = 5;              
+  int64 timestamp = 5;
 }
 
 message ColumnValue {
     bytes value = 1;
+    bool is_null = 2;
 }