presto自定义函数开发

1 Presto介绍

Presto 是 Facebook 开源的分布式查询引擎，在交互式查询任务中担当着重要的职责。随着越来越多的人开始使用 SQL 在 Presto 上分析数据，我们发现需要将一些业务逻辑开发成类似 Hive 中的 UDF，提高 SQL 使用人员的效率，同时也保证 Hive 和 Presto 环境中的 UDF 统一。

1.1 Presto函数介绍

在 Presto 中，函数大体分为三种：scalar，aggregation 和 window 类型。分别如下：

1）scalar标量函数，简单来说就是 Java 中的一个静态方法，本身没有任何状态。

2）aggregation累积状态的函数，或聚集函数，如count，avg。如果只是单节点，单机状态可以直接用一个变量存储即可，但是presto是分布式计算引擎，状态数据会在多个节点之间传输，因此状态数据需要被序列化成 Presto 的内部格式才可以被传输。

3）window 窗口函数，如同sparkSQL中的窗口函数类似

2 自定义函数实现

官网地址：https://prestodb.github.io/docs/current/develop/functions.html

2.1自定义Scalar函数的实现

2.1.1定义一个java类

1）用 @ScalarFunction 的 Annotation 标记实现业务逻辑的静态方法。

2）用 @Description 描述函数的作用，这里的内容会在 SHOW FUNCTIONS 中显示。

3）用@SqlType 标记函数的返回值类型，如返回字符串，因此是 StandardTypes.VARCHAR。

4）Java 方法的返回值必须使用 Presto 内部的序列化方式，因此字符串类型必须返回 Slice， 使用 Slices.utf8Slice 方法可以方便的将 String 类型转换成 Slice 类型

示例代码：

public class LiulishuoFunctions {

　　public static final String DATE_FORMAT = "yyyy-MM-dd";

　　@ScalarFunction
　　@Description("hive to_date function")
　　@SqlType(StandardTypes.VARCHAR)
　　public static Slice to_date(@SqlType(StandardTypes.TIMESTAMP) long input) {
　　    final DateFormat format = new SimpleDateFormat(DATE_FORMAT);
　　    return Slices.utf8Slice(format.format(new Date(input)));
　　}
　}

2.1.2 Presto插件机制

presto不能像hive那样配置自定义的udf，要采用这种插件机制实现。Presto 的插件(Plugin)机制，是 Presto 能够整合多种数据源的核心。通过实现不同的 Plugin，Presto 允许用户在不同类型的数据源之间进行 JOIN 等计算。Presto 内部的所有数据源都是通过插件机制实现， 例如 MySQL、Hive、HBase等。Presto 插件机制不仅通过加载 Connector 来实现不同数据源的访问，还通过加载 FunctionFactory 来实现 UDF 的加载。 Presto 的 Plugin 遵循 Java 中的 ServiceLoader 规范， 实现非常简单。

实现一个plugin接口如：

import com.facebook.presto.spi.Plugin;

import com.google.common.collect.ImmutableSet;
 6 
import java.util.Set;

public class PrestoFunctionsPlugin implements Plugin {
     @Override
     public Set<Class<?>> getFunctions() {
        return ImmutableSet.<Class<?>>builder()
                 .add(PvFlowStatsAggregation.class)
                 .add(AvgAggregationDemo.class)
                  .build();
    }

}

最后打包上传到指定的presto的plugin目录下，需要重启presto才能将jar中的自定义函数加载进去

2.2 自定义Aggregation函数

2.2.1实现原理步骤

Presto 把 Aggregation 函数分解成三个步骤执行：

1、input(state, data): 针对每条数据，执行 input 函数。这个过程是并行执行的，因此在每个有数据的节点都会执行，最终得到多个累积的状态数据。

2、combine(state1, state2)：将所有节点的状态数据聚合起来，多次执行，直至所有状态数据被聚合成一个最终状态，也就是 Aggregation 函数的输出结果。

3、output(final_state, out)：最终输出结果到一个 BlockBuilder。

2.2.2 具体代码实现过程

1、定义一个 Java 类，使用 @AggregationFunction 标记为 Aggregation 函数

2、使用 @InputFunction、 @CombineFunction、@OutputFunction 分别标记计算函数、合并结果函数和最终输出函数在 Plugin 处注册 Aggregation 函数

3、一个继承AccumulatorState的State接口，get和set方法

4、并使用 @AccumulatorStateMetadata 提供序列化（stateSerializerClass指定）和 Factory 类信息（stateFactoryClass指定）。自己写一个序列化类和一个工厂类。

核心代码示例：

@AggregationFunction("pv_stats")
public class PvFlowStatsAggregation {
    private PvFlowStatsAggregation() {}
    @AccumulatorStateMetadata(stateSerializerClass = PvFlowStatsStateSerializer.class,         stateFactoryClass = PvFlowStatsFactory.class)

    public interface State extends AccumulatorState {
        PvFlowStats get();
        void set(PvFlowStats value);
    }

@InputFunction
public static void input(@AggregationState State state, @SqlType(StandardTypes.BIGINT) long id,

                             @SqlType(StandardTypes.VARBINARY) Slice serialisedTree, @SqlType(StandardTypes.VARCHAR) Slice pvOrder) {
        handleDataInput(state, id, serialisedTree, pvOrder, null);
    }

@InputFunction

public static void input(@AggregationState State state, @SqlType(StandardTypes.BIGINT) long id,

                             @SqlType(StandardTypes.VARBINARY) Slice serialisedTree, @SqlType(StandardTypes.VARCHAR) Slice pvOrder,
                             @SqlType(StandardTypes.VARCHAR) Slice endUrl) {
        handleDataInput(state, id, serialisedTree, pvOrder, endUrl);
    }

    private static void handleDataInput(State state, long id, Slice serialisedTree, Slice pvOrder, Slice endUrl) {

        PvFlowStats stats = state.get();
        if (stats == null) {
            stats = new PvFlowStats();
            state.set(stats);
        }
......
}
    @CombineFunction
   public static void combine(@AggregationState State state, State other) {

        PvFlowStats input = other.get();
        PvFlowStats previous = state.get();
        if (previous == null) {
            state.set(input);
        } else {
            previous.mergeWith(input);
        }
    }
    @OutputFunction(StandardTypes.VARCHAR)
    public static void output(@AggregationState State state, BlockBuilder out) {
        PvFlowStats stats = state.get();
        if (stats == null) {
            out.appendNull();
            return;
        }
        // 统计
        Slice result = stats.statisticNextPage();
        if (result == null) {
            out.appendNull();
        } else {
            VarcharType.VARCHAR.writeSlice(out, result);
        }
    }
}

2.2.3 复杂数据类型（list,map或自定义的类）

对于复杂的类型，需要自定义序列化类和工厂类，需要自己实现类的序列化和反序列化。

下面是示例：

主类：

/*** **   id | value* ** ----+-------* **   2 | ddd* **   2 | ddd* **   1 | bbb* **   1 | bbb* **   1 | ccc* **   1 | aaa* **   1 | bbb* **   2 | aaa* **   2 | ccc* **   1 | ccc* ***   *返回* **  [{id:1,{aaa:1,ccc:2,bbb:3},{id:2,{aaa:1,ccc:1,ddd:2}]* **/*@AggregationFunction("presto_collect")
public class CollectListAggregation {

    @AccumulatorStateMetadata(stateSerializerClass = CollectListStatsSerializer.class, stateFactoryClass = CollectListStatsFactory.class)
    public interface CollectState extends AccumulatorState {
        CollectListStats get();
        void set(CollectListStats value);
    }
   @InputFunction
    public static void input(@AggregationState CollectState state, @SqlType(StandardTypes.*VARCHAR*) Slice id,@SqlType(StandardTypes.*VARCHAR*) Slice key) {
        try {
            CollectListStats stats = state.get();
            if (stats == null) {
                stats = new CollectListStats();
                state.set(stats);
            }
            int inputId = Integer.*parseInt*(id.toStringUtf8());
            String inputKey = key.toStringUtf8();
            stats.addCollectList(inputId,inputKey, 1);
       } catch (Exception e) {
            throw new RuntimeException(e+" ---------  input err");
        }
    }

    @CombineFunction
    public static void combine(@AggregationState CollectState state, CollectState otherState) {
        try {
            CollectListStats collectListStats = state.get();
            CollectListStats oCollectListStats = otherState.get();
            if(collectListStats == null) {
                state.set(oCollectListStats);
            } else {
                collectListStats.mergeWith(oCollectListStats);
            }
        }catch (Exception e) {
            throw new RuntimeException(e+" --------- combine err");
        }
    }

    @OutputFunction(StandardTypes.*VARCHAR*)
    public static void output(@AggregationState CollectState state, BlockBuilder out) {
        try {
            CollectListStats stats = state.get();
            if (stats == null) {
                out.appendNull();
                return;
            }
            // 统计
            Slice result = stats.getCollectResult();
            if (result == null) {
                out.appendNull();
            } else {
                VarcharType.*VARCHAR*.writeSlice(out, result);
            }
        } catch (Exception e) {
            throw new RuntimeException(e+" -------- output err");
        }
    }
}

主类实现的比较简单，input,combine,output即可

存放数据的类：此类需要实现数据的序列化和反序列化，这是最关键和比较麻烦的地方，贴一个例子，关键在于需要自己控制存储空间以及数据的顺序，和读取的时候按照一定顺序读取。对于字符要先存储长度，然后是字节，读取则先读取字符长度，然后读取这么长的数据，最后转化为字符

public class CollectListStats {
    private static final int *INSTANCE_SIZE* = ClassLayout.*parseClass*(CollectListStats.class).instanceSize();
    //<id,<key,value>>
    private Map<Integer,Map<String,Integer>> collectContainer = new HashMap<>();
    private long contentEstimatedSize = 0;
    private int keyByteLen = 0;
    private int keyListLen = 0;
    CollectListStats() {
    }
    CollectListStats(Slice serialized) {
        deserialize(serialized);
    }
   void addCollectList(Integer id, String key, int value) {
        if (collectContainer.containsKey(id)) {
            Map<String, Integer> tmpMap = collectContainer.get(id);
            if (tmpMap.containsKey(key)) {
                tmpMap.put(key, tmpMap.get(key)+value);
            }else{
                tmpMap.put(key,value);
                contentEstimatedSize += ( key.getBytes().length + SizeOf.*SIZE_OF_INT*);
                keyByteLen += key.getBytes().length;
                keyListLen++;
            }
        } else {
            Map<String,Integer> tmpMap = new HashMap<String,Integer>();
            tmpMap.put(key, value);
            keyByteLen += key.getBytes().length;
            keyListLen++;
            collectContainer.put(id, tmpMap);
            contentEstimatedSize += SizeOf.*SIZE_OF_INT*;
        }
    }
    //[{id:1,{"aaa":3,"fadf":6},{}]
    Slice getCollectResult() {
        Slice jsonSlice = null;
        try {
            StringBuilder jsonStr = new StringBuilder();
            jsonStr.append("[");
            int collectLength = collectContainer.entrySet().size();
            for (Map.Entry<Integer, Map<String, Integer>> mapEntry : collectContainer.entrySet()) {
                Integer id = mapEntry.getKey();
                Map<String, Integer> vMap = mapEntry.getValue();
                jsonStr.append("{id:").append(id).append(",{");
                int vLength = vMap.entrySet().size();
                for (Map.Entry<String, Integer> vEntry : vMap.entrySet()) {
                    String key = vEntry.getKey();
                    Integer value = vEntry.getValue();
                    jsonStr.append(key).append(":").append(value);
                    vLength--;
                    if (vLength != 0) {
                        jsonStr.append(",");
                    }
                }
                jsonStr.append("}");
                collectLength--;
                if (collectLength != 0) {
                    jsonStr.append(",");
                }
            }
            jsonStr.append("]");
            jsonSlice = Slices.*utf8Slice*(jsonStr.toString());
        } catch (Exception e) {
            throw new RuntimeException(e+" ---------- get CollectResult err");
        }
        return jsonSlice;
    }
    public void deserialize(Slice serialized) {
        try {
            SliceInput input = serialized.getInput();
            //外层map的长度
            int collectStatsEntrySize = input.readInt();
            for (int collectCnt = 0; collectCnt < collectStatsEntrySize; collectCnt++) {

                int id = input.readInt();
                int keyEntrySize = input.readInt();
                for (int idCnt = 0; idCnt < keyEntrySize; idCnt++) {
                    int keyBytesLen = input.readInt();
                    byte[] keyBytes = new byte[keyBytesLen];
                    for (int byteIdx = 0; byteIdx < keyBytesLen; byteIdx++) {
                        keyBytes[byteIdx] = input.readByte();
                    }
                    String key = new String(keyBytes);
                    int value = input.readInt();
                    addCollectList(id, key, value);
                }
            }
        } catch (Exception e) {
            throw new RuntimeException(e+" ----- deserialize err");
        }
    }

    public Slice serialize() {
        SliceOutput builder = null;
        int requiredBytes =                                                 //对应 SliceOutput builder append的内容所占用的空间
                SizeOf.*SIZE_OF_INT* * 3                                      //id entry数目，id数值，key Entry数目
                + keyListLen * SizeOf.*SIZE_OF_INT*                           //key bytes长度
                + keyByteLen                                                //key byte总长度
                + keyListLen * SizeOf.*SIZE_OF_INT*;                          //value
        try {
            // 序列化
            builder = Slices.*allocate*(requiredBytes).getOutput();
            for (Map.Entry<Integer,Map<String, Integer>> entry : collectContainer.entrySet()) {
                //id个数
                builder.appendInt(collectContainer.entrySet().size());
                //id 数值
                builder.appendInt(entry.getKey());
                Map<String, Integer> kMap = entry.getValue();
                builder.appendInt(kMap.entrySet().size());
                for (Map.Entry<String, Integer> vEntry : kMap.entrySet()) {
                    byte[] keyBytes = vEntry.getKey().getBytes();
                    builder.appendInt(keyBytes.length);
                    builder.appendBytes(keyBytes);
                    builder.appendInt(vEntry.getValue());
                }
            }
            return builder.getUnderlyingSlice();
        } catch (Exception e) {
            throw new RuntimeException(e+" ---- serialize err  requiredBytes = " + requiredBytes + " keyByteLen= " + keyByteLen + " keyListLen = " + keyListLen);
       }
    }
    long estimatedInMemorySize() {
        return *INSTANCE_SIZE* + contentEstimatedSize;
    }
    void mergeWith(CollectListStats other) {
        if (other == null) {
            return;
        }
       for (Map.Entry<Integer,Map<String, Integer>> cEntry : other.collectContainer.entrySet()) {
            Integer id = cEntry.getKey();
            Map<String, Integer> kMap = cEntry.getValue();
            for (Map.Entry<String, Integer> kEntry : kMap.entrySet()) {
                addCollectList(id, kEntry.getKey(), kEntry.getValue());
            }
        }
    }
}

序列化类：

public class CollectListStatsSerializer implements AccumulatorStateSerializer<CollectListAggregation.CollectState> {
    @Override
    public Type getSerializedType() {
        return *VARBINARY*;
    }
    @Override
    public void serialize(CollectListAggregation.CollectState state, BlockBuilder out) {
        if (state.get() == null) {
            out.appendNull();
        } else {
            *VARBINARY*.writeSlice(out, state.get().serialize());
        }
    }
    @Override
    public void deserialize(Block block, int index, CollectListAggregation.CollectState state) {
        state.set(new CollectListStats(*VARBINARY*.getSlice(block, index)));
    }
}

工厂类：

/*** **/*public class CollectListStatsFactory implements AccumulatorStateFactory<CollectListAggregation.CollectState> {
    @Override
    public CollectListAggregation.CollectState createSingleState() {
        return new SingleState();
    }
    @Override
    public Class<? extends CollectListAggregation.CollectState> getSingleStateClass() {
        return SingleState.class;
    }
    @Override
    public CollectListAggregation.CollectState createGroupedState() {
        return new GroupState();
    }
    @Override
    public Class<? extends CollectListAggregation.CollectState> getGroupedStateClass() {
        return GroupState.class;
    }
    public static class GroupState implements GroupedAccumulatorState, CollectListAggregation.CollectState {
        private final ObjectBigArray<CollectListStats> collectStatsList = new ObjectBigArray<>();
        private long size;
        private long groupId;
        @Override
        public void setGroupId(long groupId) {
            this.groupId = groupId;
        }
        @Override
        public void ensureCapacity(long size) {
            collectStatsList.ensureCapacity(size);
        }
        @Override
        public CollectListStats get() {
            return collectStatsList.get(groupId);
        }
        @Override
        public void set(CollectListStats value) {
            CollectListStats previous = get();
            if (previous != null) {
                size -= previous.estimatedInMemorySize();
            }
            collectStatsList.set(groupId, value);
            size += value.estimatedInMemorySize();
        }
        @Override
        public long getEstimatedSize() {
            return size + collectStatsList.sizeOf();
        }
    }
    public static class SingleState implements CollectListAggregation.CollectState{
        private CollectListStats stats;
        @Override
        public CollectListStats get() {
            return stats;
        }
        @Override
        public void set(CollectListStats value) {
            stats = value;
        }
        @Override
        public long getEstimatedSize() {
            if (stats == null) {
                return 0;
            }
            return stats.estimatedInMemorySize();
        }
    }
}

2.2.4 采用Slice可以有效提高性能

使用Slice进行内存操作，Slice使用Unsafe#copyMemory实现了高效的内存拷贝

不过使用Slice就需要手动控制存储的数据，记录数据的容量，长度，扩容等等。

public class RouteUserAggregationBase {

    //...... 其他定义的静态变量
    /**
     * Slice State
     * 中间数据 Buffer
     */
    public interface SliceState extends AccumulatorState {
        Slice getSlice();
    
        void setSlice(Slice slice);
    }
}

@AggregationFunction("函数名")
public class RouteUserGroupAggregation extends RouteUserAggregationBase {

    /** 缓存 Buffer Body 的初始字节容量 **/
    private static final int STORED_DATA_BODY_INIT_BYTE_SIZE = 64;

    /** 缓存 Buffer 头部元信息定义 **/
    private static int VALUES_OFFSET_HEADER_BYTE_LEN = 0;
    private static int VALUES_OFFSET_BODY_BYTE_SIZE = 4;
    private static int VALUES_OFFSET_BODY_BYTE_USED = 8;

    private static int VALUES_OFFSET_CONTAIN_TARGET_EVENT = 12;

    private static int VALUES_OFFSET_TARGET_EVENT_TYPE = 13;
    private static int VALUES_OFFSET_ROUTE_INTERVAL = 17;
    private static int VALUES_OFFSET_TARGET_EVENT_LEN = 21;
    private static int VALUES_OFFSET_TARGET_EVENT_BYTES = 25;

    @InputFunction
    public static void input(SliceState state,
                             //目标事件
                             @SqlType(StandardTypes.VARCHAR) Slice targetEvent,
                             //目标事件类型
                             @SqlType(StandardTypes.BIGINT) long targetType,
                             //事件间隔
                             @SqlType(StandardTypes.BIGINT) long eventInterval,
                             //当前事件名
                             @SqlType(StandardTypes.VARCHAR) Slice currEvent,
                             //当前事件时间
                             @SqlType(StandardTypes.BIGINT) long eventTime) {

        handleInput(state, targetEvent, (int) targetType, (int) eventInterval, currEvent, (int) eventTime, null, null);
    }

    private static void handleInput(SliceState state, Slice targetEvent, int targetType, int eventInterval, Slice currEvent, int eventTime, Slice groupByEvent, Slice groupByProp) {
        // 获取缓存的数据
        Slice storedData = state.getSlice();

        // 初始化缓存的元信息 不会变化的值，如：目标事件，目标类型，时间间隔
        if (storedData == null) {
            /*
                    Header byte大小
                    Body 总字节大小
                    Body 已使用字节大小
                    是否包含目标事件
                    目标事件类型
                    事件时间间隔
             */
            int headerByteLen = SizeOf.SIZE_OF_INT
                    + SizeOf.SIZE_OF_INT
                    + SizeOf.SIZE_OF_INT
                    + SizeOf.SIZE_OF_BYTE
                    + SizeOf.SIZE_OF_INT
                    + SizeOf.SIZE_OF_INT
                    ;
            int targetLength = SizeOf.SIZE_OF_INT + targetEvent.length();
            headerByteLen += targetLength;

            storedData = Slices.allocate(headerByteLen + STORED_DATA_BODY_INIT_BYTE_SIZE);
            storedData.setInt(VALUES_OFFSET_HEADER_BYTE_LEN, headerByteLen);
            storedData.setInt(VALUES_OFFSET_BODY_BYTE_SIZE, STORED_DATA_BODY_INIT_BYTE_SIZE);
            storedData.setInt(VALUES_OFFSET_BODY_BYTE_USED, 0);
            //是否包含目标事件
            storedData.setByte(VALUES_OFFSET_CONTAIN_TARGET_EVENT, 0);
            //缓存 不变的参数
            storedData.setInt(VALUES_OFFSET_TARGET_EVENT_TYPE, targetType);
            storedData.setInt(VALUES_OFFSET_ROUTE_INTERVAL, eventInterval);

            storedData.setInt(VALUES_OFFSET_TARGET_EVENT_LEN, targetEvent.length());
            storedData.setBytes(VALUES_OFFSET_TARGET_EVENT_BYTES, targetEvent);
        }

        int headerByteLen = storedData.getInt(VALUES_OFFSET_HEADER_BYTE_LEN);
        int bodyByteSize = storedData.getInt(VALUES_OFFSET_BODY_BYTE_SIZE);
        int bodyByteUsed = storedData.getInt(VALUES_OFFSET_BODY_BYTE_USED);

        // 标记包含目标事件
        if (currEvent.toStringUtf8().equals(targetEvent.toStringUtf8())) {
            storedData.setByte(VALUES_OFFSET_CONTAIN_TARGET_EVENT, 1);
        }
        //直接判断，如果存在分组，判断当前事件就是分组事件，那么直接将分组值和事件拼接在一起
        if (groupByEvent != null && groupByEvent.toStringUtf8().equals(currEvent.toStringUtf8())) {
            String newEventKey = currEvent.toStringUtf8() + EVENT_CONCAT_GROUP_VALUE + groupByProp.toStringUtf8();
            currEvent = Slices.utf8Slice(newEventKey);
        }

        //扩展的长度，eventTime int , current length的int bytes内容
        int entryByteLen = SizeOf.SIZE_OF_INT * 2 + currEvent.length();
        if (bodyByteUsed + entryByteLen > bodyByteSize) {
            // 扩容 byteSize * 2
            int newBodyByteSize = bodyByteSize * 2;
            Slice newStoredData = Slices.allocate(headerByteLen + newBodyByteSize);
            //将storeData的数据copy到new的Slice中，然后重新设置容量
            newStoredData.setBytes(0, storedData.getBytes());
            newStoredData.setInt(VALUES_OFFSET_BODY_BYTE_SIZE, newBodyByteSize);
            storedData = newStoredData;
        }
        //写入位置的定位
        int writePos = headerByteLen + bodyByteUsed;
        storedData.setInt(writePos, entryByteLen);
        writePos += SizeOf.SIZE_OF_INT;
        storedData.setInt(writePos, eventTime);
        writePos += SizeOf.SIZE_OF_INT;
        storedData.setBytes(writePos, currEvent);
        storedData.setInt(VALUES_OFFSET_BODY_BYTE_USED, bodyByteUsed + entryByteLen);
        // 更新缓存的数据
        state.setSlice(storedData);
    }


    @CombineFunction
    public static void combine(SliceState state, SliceState other) {
        // 获取缓存的数据
        Slice storedData = state.getSlice();
        Slice otherStoredData = other.getSlice();

        // 合并缓存
        if (storedData == null) {
            state.setSlice(otherStoredData);
        } else {
            int headerByteLen = storedData.getInt(VALUES_OFFSET_HEADER_BYTE_LEN);
            int bodyByteSize = storedData.getInt(VALUES_OFFSET_BODY_BYTE_SIZE);
            int bodyByteUsed = storedData.getInt(VALUES_OFFSET_BODY_BYTE_USED);
            int otherHeaderByteLen = otherStoredData.getInt(VALUES_OFFSET_HEADER_BYTE_LEN);
            int otherBodyByteSize = otherStoredData.getInt(VALUES_OFFSET_BODY_BYTE_SIZE);
            int otherBodyByteUsed = otherStoredData.getInt(VALUES_OFFSET_BODY_BYTE_USED);
            byte containTargetEvent = 0;
            if (storedData.getByte(VALUES_OFFSET_CONTAIN_TARGET_EVENT) == 1 || otherStoredData.getByte(VALUES_OFFSET_CONTAIN_TARGET_EVENT) == 1) {
                containTargetEvent = 1;
            }
            Slice finalStoredData;
            int finalBodyByteUsed = bodyByteUsed + otherBodyByteUsed;
            if (bodyByteSize >= finalBodyByteUsed) {
                // 左容量足够  这里只copy header之外的数据，就是当前事件和time
                storedData.setBytes(headerByteLen + bodyByteUsed, otherStoredData, otherHeaderByteLen, otherBodyByteUsed);
                storedData.setInt(VALUES_OFFSET_BODY_BYTE_USED, finalBodyByteUsed);
                finalStoredData = storedData;
            } else if (otherBodyByteSize >= finalBodyByteUsed) {
                // 右容量足够
                otherStoredData.setBytes(otherHeaderByteLen + otherBodyByteUsed, storedData, headerByteLen, bodyByteUsed);
                otherStoredData.setInt(VALUES_OFFSET_BODY_BYTE_USED, finalBodyByteUsed);
                finalStoredData = otherStoredData;
            } else {
                // 扩容
                int newBodyByteSize = bodyByteSize;
                while (newBodyByteSize < finalBodyByteUsed) {
                    newBodyByteSize *= 2;
                }
                Slice newStoredData = Slices.allocate(headerByteLen + newBodyByteSize);
                newStoredData.setBytes(VALUES_OFFSET_HEADER_BYTE_LEN, storedData.getBytes());
                newStoredData.setInt(VALUES_OFFSET_BODY_BYTE_SIZE, newBodyByteSize);
                storedData = newStoredData;

                storedData.setBytes(headerByteLen + bodyByteUsed, otherStoredData, otherHeaderByteLen, otherBodyByteUsed);
                storedData.setInt(VALUES_OFFSET_BODY_BYTE_USED, finalBodyByteUsed);
                finalStoredData = storedData;
            }
            // 是否包含目标事件
            finalStoredData.setByte(VALUES_OFFSET_CONTAIN_TARGET_EVENT, containTargetEvent);
            state.setSlice(finalStoredData);
        }
    }

    @OutputFunction(StandardTypes.VARCHAR)
    public static void output(@AggregationState SliceState state, BlockBuilder out) {
        // 获取缓存数据
        Slice storedData = state.getSlice();

        // 数据为空，或者没有起始事件
        if ((storedData == null) || (storedData.getByte(VALUES_OFFSET_CONTAIN_TARGET_EVENT) == 0)) {
            out.appendNull();
            return;
        }
        //匹配
        Slice makeRoute = makeRoute(storedData);
        if (makeRoute == null) {
            out.appendNull();
        } else {
            VarcharType.VARCHAR.writeSlice(out, makeRoute);
        }
    }

    private static Slice makeRoute(Slice storedData) {
        // 获取 Header 信息
        int interval = storedData.getInt(VALUES_OFFSET_ROUTE_INTERVAL);
        int targetType = storedData.getInt(VALUES_OFFSET_TARGET_EVENT_TYPE);
        int targetLength = storedData.getInt(VALUES_OFFSET_TARGET_EVENT_LEN);
        String targetEvent = new String(storedData.getBytes(VALUES_OFFSET_TARGET_EVENT_BYTES, targetLength));
        List<Slice> timeEventSeries = new ArrayList<>();
        int headerByteLen = storedData.getInt(VALUES_OFFSET_HEADER_BYTE_LEN);
        int bodyByteUsed = storedData.getInt(VALUES_OFFSET_BODY_BYTE_USED);
        int bound = headerByteLen + bodyByteUsed;
        int idx = headerByteLen;
        while (idx < bound) {
            //获取每个事件数据 time，事件名
            int entryByteLen = storedData.getInt(idx);
            Slice entry = storedData.slice(idx + SizeOf.SIZE_OF_INT, entryByteLen - SizeOf.SIZE_OF_INT);
            idx += entryByteLen;
            timeEventSeries.add(entry);
        }
        //处理逻辑
        ......
        // 构造返回结果
        Slice result = null;
        if (routes.size() > 0) {
            for (String route : routes) {
                Slice routeSlice = Slices.utf8Slice(route);
                Slice routeInfo = Slices.allocate(SizeOf.SIZE_OF_INT + routeSlice.length());
                routeInfo.setInt(0, routeSlice.length());
                routeInfo.setBytes(4, routeSlice);
                if (result == null) {
                    result = routeInfo;
                } else {
                    Slice newSlice = Slices.allocate(result.length() + routeInfo.length());
                    newSlice.setBytes(0, result);
                    newSlice.setBytes(result.length(), routeInfo, 0, routeInfo.length());
                    result = newSlice;
                }
            }
        }
        return result;
    }
}