/**
 * Needed to deserialize an instance of this {@link GroupDataSummary} from a {@link Memory}.
 *
 * @param serializedSummary The serialized summary as a {@link Memory} object.
 * @return A {@link DeserializeResult} representing the deserialized summary.
 */
public static DeserializeResult<GroupDataSummary> fromMemory(Memory serializedSummary) {
    byte initialized = serializedSummary.getByte(INITIALIZED_POSITION);
    int size = serializedSummary.getInt(SIZE_POSITION);

    byte[] data = new byte[size];
    serializedSummary.getByteArray(DATA_POSITION, data, 0, size);
    GroupData deserializedData = SerializerDeserializer.fromBytes(data);

    GroupDataSummary deserialized = new GroupDataSummary();
    deserialized.initialized = initialized != 0;
    deserialized.data = deserializedData;

    // Size read is the size of size and the byte in bytes (DATA_POSITION) plus the size of the data (size)
    return new DeserializeResult<>(deserialized, size + DATA_POSITION);
}
 

@Test
public void testSummaryDeserialization() {
    GroupDataSummaryFactory factory = new GroupDataSummaryFactory();
    GroupDataSummary summary = new GroupDataSummary();

    CachingGroupData data = sampleSumGroupData(30.0);
    summary.update(data);

    byte[] serializedSummary = summary.toByteArray();
    Memory memory = new NativeMemory(serializedSummary);

    DeserializeResult result = factory.summaryFromMemory(memory);
    GroupDataSummary deserialized = (GroupDataSummary) result.getObject();

    Assert.assertNotNull(deserialized);

    BulletRecord actual = deserialized.getData().getAsBulletRecord(emptyMap(), new BulletConfig().getBulletRecordProvider());
    BulletRecord expected = RecordBox.get().add("fieldA", "bar").add("sum", 30.0).getRecord();
    Assert.assertTrue(actual.equals(expected));
}
 

@Test
public void testSerializationNullData() {
    GroupDataSummary summary = new GroupDataSummary();
    summary.setData(null);

    byte[] serialized = summary.toByteArray();
    Assert.assertNotNull(serialized);

    Memory memory = new NativeMemory(serialized);

    byte initializedByte = memory.getByte(GroupDataSummary.INITIALIZED_POSITION);
    int length = memory.getInt(GroupDataSummary.SIZE_POSITION);
    byte[] dataSerialized = new byte[length];
    memory.getByteArray(GroupDataSummary.DATA_POSITION, dataSerialized, 0, length);
    CachingGroupData data = SerializerDeserializer.fromBytes(dataSerialized);

    Assert.assertTrue(initializedByte == 0);
    Assert.assertTrue(length > 0);
    Assert.assertNull(data);
}
 

@Test
public void testDeserialization() {
    List<String> groups = asList("foo", "bar", "baz");
    List<GroupOperation.GroupOperationType> operations = asList(COUNT, MAX, MIN);
    CachingGroupData data = new CachingGroupData(makeGroups(groups), null, makeMetrics(operations));

    byte[] serializedData = SerializerDeserializer.toBytes(data);
    int serializedDataLength = serializedData.length;

    Memory memory = new NativeMemory(new byte[GroupDataSummary.DATA_POSITION + serializedDataLength]);
    memory.putByte(GroupDataSummary.INITIALIZED_POSITION, (byte) 1);
    memory.putInt(GroupDataSummary.SIZE_POSITION, serializedDataLength);
    memory.putByteArray(GroupDataSummary.DATA_POSITION, serializedData, 0, serializedDataLength);

    DeserializeResult<GroupDataSummary> deserialized = GroupDataSummary.fromMemory(memory);

    Assert.assertNotNull(deserialized);
    Assert.assertEquals(deserialized.getSize(), GroupDataSummary.DATA_POSITION + serializedDataLength);

    Assert.assertTrue(deserialized.getObject().isInitialized());

    GroupData deserializedData = deserialized.getObject().getData();
    Assert.assertEquals(deserializedData.groupFields, data.groupFields);
    Assert.assertEquals(deserializedData.metrics, data.metrics);
}
 

@Override
public byte[] toByteArray() {
    byte[] groupData = SerializerDeserializer.toBytes(data);
    int length = groupData.length;

    // Create a new ByteBuffer to hold a byte, an integer and the data in bytes
    byte[] serialized = new byte[DATA_POSITION + length];
    Memory memory = new NativeMemory(serialized);
    memory.putByte(INITIALIZED_POSITION, (byte) (initialized ? 1 : 0));
    memory.putInt(SIZE_POSITION, length);
    memory.putByteArray(DATA_POSITION, groupData, 0, length);
    return serialized;
}
 

@Test
public void testDeserialization() {
    GroupDataSummaryFactory factory = new GroupDataSummaryFactory();
    byte[] serialized = factory.toByteArray();
    Memory memory = new NativeMemory(serialized);

    DeserializeResult<GroupDataSummaryFactory> deserialized = GroupDataSummaryFactory.fromMemory(memory);
    Assert.assertNotNull(deserialized.getObject());
    Assert.assertEquals(deserialized.getSize(), GroupDataSummaryFactory.SERIALIZED_SIZE);
}
 

@Test
public void testStaticDeserialization() {
    GroupDataSummaryFactory factory = new GroupDataSummaryFactory();
    byte[] serialized = new byte[10];
    Memory memory = new NativeMemory(serialized);

    // This is testing that not matter what you send it, it will deserialize it
    DeserializeResult<GroupDataSummaryFactory> deserialized = GroupDataSummaryFactory.fromMemory(memory);
    Assert.assertNotNull(deserialized.getObject());
    Assert.assertEquals(deserialized.getSize(), GroupDataSummaryFactory.SERIALIZED_SIZE);

    deserialized = GroupDataSummaryFactory.fromMemory(null);
    Assert.assertNotNull(deserialized.getObject());
    Assert.assertEquals(deserialized.getSize(), GroupDataSummaryFactory.SERIALIZED_SIZE);
}
 

@Test
public void testEmptySummaryDeserialization() {
    GroupDataSummaryFactory factory = new GroupDataSummaryFactory();
    GroupDataSummary summary = new GroupDataSummary();

    byte[] serializedSummary = summary.toByteArray();
    Memory memory = new NativeMemory(serializedSummary);

    DeserializeResult result = factory.summaryFromMemory(memory);
    GroupDataSummary deserialized = (GroupDataSummary) result.getObject();

    Assert.assertNotNull(deserialized);
    Assert.assertNull(deserialized.getData());
}
 

@Test
public void testSerialization() {
    List<String> groups = asList("foo", "bar", "baz");
    List<GroupOperation.GroupOperationType> operations = asList(COUNT, MAX, MIN);
    CachingGroupData data = new CachingGroupData(makeGroups(groups), null, makeMetrics(operations));
    data.setCachedRecord(RecordBox.get().getRecord());

    GroupDataSummary summary = new GroupDataSummary();
    summary.update(data);

    byte[] serialized = summary.toByteArray();
    Assert.assertNotNull(serialized);

    Memory memory = new NativeMemory(serialized);

    byte initializedByte = memory.getByte(GroupDataSummary.INITIALIZED_POSITION);
    int length = memory.getInt(GroupDataSummary.SIZE_POSITION);
    byte[] dataSerialized = new byte[length];
    memory.getByteArray(GroupDataSummary.DATA_POSITION, dataSerialized, 0, length);
    CachingGroupData deserialized = SerializerDeserializer.fromBytes(dataSerialized);

    Assert.assertTrue(initializedByte == 1);
    Assert.assertTrue(length > 0);

    Assert.assertNotNull(deserialized);
    Assert.assertNull(deserialized.getCachedRecord());

    BulletRecord actual = deserialized.getAsBulletRecord(emptyMap(), provider);
    BulletRecord expected = RecordBox.get().add("field_0", "foo").add("field_1", "bar").add("field_2", "baz")
                                           .add("COUNT_metric_0", 1L).addNull("MAX_metric_1").addNull("MIN_metric_2")
                                           .getRecord();

    Assert.assertTrue(actual.equals(expected));
}
 

@Override
public void accumulate(final long memoryAddr, final int count) {
  final long maxAddr = memoryAddr + count * 4;
  final long incomingBit = getInput().getValidityBufferAddress();
  final long incomingValue =  getInput().getDataBufferAddress();

  int incomingIndex = 0;

  final ArrowBuf inputOffsetBuf = getInput().getOffsetBuffer();
  final ArrowBuf inputBuf = getInput().getDataBuffer();

  for(long ordinalAddr = memoryAddr; ordinalAddr < maxAddr; ordinalAddr += 4, incomingIndex++){
    final int bitVal = (PlatformDependent.getByte(incomingBit + ((incomingIndex >>> 3))) >>> (incomingIndex & 7)) & 1;

    //incoming record is null, skip it
    if (bitVal == 0) {
      continue;
    }

    //get the proper chunk from the ordinal
    final int tableIndex = PlatformDependent.getInt(ordinalAddr);
    //System.out.println("record idx: " + incomingIndex + " ordinal: " + tableIndex);
    final int chunkIndex = tableIndex >>> LBlockHashTableNoSpill.BITS_IN_CHUNK;
    final int chunkOffset = tableIndex & LBlockHashTableNoSpill.CHUNK_OFFSET_MASK;

    final HllAccumHolder ah =  this.accumulators[chunkIndex];
    final HllSketch sketch = ah.getAccums()[chunkOffset];

    //get the offset of incoming record
    final int startOffset = inputOffsetBuf.getInt(incomingIndex * BaseVariableWidthVector.OFFSET_WIDTH);
    final int endOffset = inputOffsetBuf.getInt((incomingIndex + 1) * BaseVariableWidthVector.OFFSET_WIDTH);
    final int len = endOffset - startOffset;
    final ByteBuffer buffer = inputBuf.nioBuffer(startOffset, len);

    //apply the update
    sketch.update(Memory.wrap(buffer), 0, len);
  } //for
}
 

@Override
public void accumulate(final long memoryAddr, final int count) {
  final long maxAddr = memoryAddr + count * WIDTH_ORDINAL;
  FieldVector inputVector = getInput();
  final long incomingBit = inputVector.getValidityBufferAddress();
  final long incomingValue = inputVector.getDataBufferAddress();
  final int scale = ((DecimalVector) inputVector).getScale();

  int incomingIndex = 0;
  for (long ordinalAddr = memoryAddr; ordinalAddr < maxAddr; ordinalAddr += WIDTH_ORDINAL, incomingIndex++) {
    /* get the corresponding data from input vector -- source data for accumulation */
    final int bitVal = (PlatformDependent.getByte(incomingBit + ((incomingIndex >>> 3))) >>> (incomingIndex & 7)) & 1;
    // without if we would need to do a multiply which is slow.
    if (bitVal == 0) {
      continue;
    }

    final ByteBuffer buffer = inputVector.getDataBuffer().nioBuffer(incomingIndex * WIDTH_INPUT, WIDTH_INPUT);

    final int tableIndex = PlatformDependent.getInt(ordinalAddr);
    int chunkIndex = tableIndex >>> LBlockHashTableNoSpill.BITS_IN_CHUNK;
    int chunkOffset = tableIndex & LBlockHashTableNoSpill.CHUNK_OFFSET_MASK;

    final HllAccumHolder ah =  this.accumulators[chunkIndex];
    final HllSketch sketch = ah.getAccums()[chunkOffset];

    sketch.update(Memory.wrap(buffer), 0, WIDTH_INPUT);
  } //for
}
 

@Override
public void accumulate(final long memoryAddr, final int count) {
  final long maxAddr = memoryAddr + count * 4;
  final long incomingBit = getInput().getValidityBufferAddress();
  final long incomingValue =  getInput().getDataBufferAddress();

  int incomingIndex = 0;

  final ArrowBuf inputOffsetBuf = getInput().getOffsetBuffer();
  final ArrowBuf inputBuf = getInput().getDataBuffer();

  for(long ordinalAddr = memoryAddr; ordinalAddr < maxAddr; ordinalAddr += 4, incomingIndex++){
    final int bitVal = (PlatformDependent.getByte(incomingBit + ((incomingIndex >>> 3))) >>> (incomingIndex & 7)) & 1;

    //incoming record is null, skip it
    if (bitVal == 0) {
      continue;
    }

    //get the proper chunk from the ordinal
    final int tableIndex = PlatformDependent.getInt(ordinalAddr);
    //System.out.println("record idx: " + incomingIndex + " ordinal: " + tableIndex);
    final int chunkIndex = tableIndex >>> LBlockHashTableNoSpill.BITS_IN_CHUNK;
    final int chunkOffset = tableIndex & LBlockHashTableNoSpill.CHUNK_OFFSET_MASK;

    //get the offset of incoming record
    final int startOffset = inputOffsetBuf.getInt(incomingIndex * BaseVariableWidthVector.OFFSET_WIDTH);
    final int endOffset = inputOffsetBuf.getInt((incomingIndex + 1) * BaseVariableWidthVector.OFFSET_WIDTH);
    final int len = endOffset - startOffset;
    ByteBuffer buffer = inputBuf.nioBuffer(startOffset, len);

    //apply the update
    final HllUnionAccumHolder ah =  this.accumulators[chunkIndex];
    final Union unionSketch = ah.getAccums()[chunkOffset];
    final HllSketch sketch = HllSketch.wrap(Memory.wrap(buffer));
    unionSketch.update(sketch);
  } //for
}
 

@Override
public KllFloatsSketch deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) {
    return KllFloatsSketch.heapify(Memory.wrap(Base64.getDecoder().decode(json.getAsString())));
}
 

@Override
public DeserializeResult summaryFromMemory(Memory serializedSummary) {
    return GroupDataSummary.fromMemory(serializedSummary);
}
 

@Test(expectedExceptions = NegativeArraySizeException.class)
public void testBadMemoryDeserialization() {
    Memory badMemory = new NativeMemory(new byte[16]);
    badMemory.putInt(GroupDataSummary.SIZE_POSITION, -1);
    GroupDataSummary.fromMemory(badMemory);
}
 

public void addHll(ArrowBuf buf, int start, int end) {
  final int len = end - start;
  HllSketch sketch = HllSketch.wrap(Memory.wrap(buf.nioBuffer(start, len)));
  union.update(sketch);
}
 

public void addBytes(final ArrowBuf buf, final int start, final int end) {
  final int len = end - start;
  ByteBuffer buffer = buf.nioBuffer(start, len);
  Memory.wrap(buffer);
  sketch.update(Memory.wrap(buffer), 0, len);
}
 

public static long getEstimate(ArrowBuf buf, int start, int end) {
  return (long) HllSketch.wrap(Memory.wrap(buf.nioBuffer(start, end - start))).getEstimate();
}
 

/**
 * Needed to deserialize an instance of this {@link GroupDataSummaryFactory} from a {@link Memory}.
 *
 * @param summaryFactory The serialized summary factory.
 * @return A {@link DeserializeResult} representing the deserialized summary factory.
 */
public static DeserializeResult<GroupDataSummaryFactory> fromMemory(Memory summaryFactory) {
    // This has no state so it does not use the Memory
    return new DeserializeResult<>(new GroupDataSummaryFactory(), SERIALIZED_SIZE);
}