置顶通知

由于评论从畅言云评迁移到waline，导致评论数据丢失。非本人删除，各位大佬见谅！！！

Trino Sql优化:Removing redundant ORDER BY

一些SQL构造（如ORDER BY）在许多情况下不会影响查询结果，并且会产生影响性能的负面影响(查询中的每个ORDER BY子句都代表一个排序执行计划)。如果用户无意中在没有效果的地方使用ORDER BY，可能会导致严重的性能下降和资源浪费。

sql规范(ISO 9075 Part 2)中说明:

一个<query expression>可以包含一个可选的<order by clause>。
<query expression>z中行的顺序仅由<query expression>直接包含的<order by clause>指定。

上述规范意味着，查询引擎可以自由地忽略任何不适合上下文的ORDER BY子句。

Trino源码学习-自定义函数

Presto Functions 并不能像 Hive UDF 一样动态加载，需要根据 Function 的类型，实现 Presto 内部定义的不同接口，在 Presto 服务启动时进行注册，然后才能在 SQL 执行时进行调用。

1. 函数定义

Presto 内部将 Functions 分为以下三大类：

• Scalar Function，即标量函数。将传递给它的一个或者多个参数值，进行计算后，返回一个确定类型的标量值。
• Aggregation Function，即聚合函数。计算从列中取得的值，返回一个单一的值。
• Window Function，即开窗函数。计算从分组列中取得的值，并返回多个值。

对于不同类型的函数，需要遵循不同的规则进行实现。

1.1 标量函数

Presto 使用注解框架来实现标量函数，标量函数分别需要定义函数名称、输入参数类型和返回结果类型。下面介绍几种开发标量函数常用的注解：

• @ScalarFunction：用于声明标量函数的名称和别名
• @Description：用于生成函数的功能描述
• @SqlType：用于声明函数的返回类型和参数类型
• @TypeParameter：用于声明类型变量，它所声明的类型变量可以用于函数的返回类型和参数类型，框架在运行时会自动将变量与具体的类型进行绑定
• @SqlNullable：用于表示函数参数或返回结果可能为NULL。如果方法的参数不使用此注解，当函数参数包含NULL时，则该函数不会被调用，框架自动返回结果NULL。当 Java 代码中用于实现函数的方法的返回值为包装类型时，必须要在实现方法上加上该注解，且该注解无法用于 Java 基础类型

下面用一个简单的is_null函数来具体说明如何使用以上注解进行标量函数开发。

1
2
3
4
5
6
7
8
9
10

public class ExampleIsNullFunction
{
    @ScalarFunction(value = "is_null", alias = "isnull")
    @Description("Returns TRUE if the argument is NULL")
    @SqlType(StandardTypes.BOOLEAN)
    public static boolean isNull(@SqlNullable @SqlType(StandardTypes.VARCHAR) Slice string)
    {
        return (string == null);
    }
}

以上代码实现的is_null函数功能为：判断传入的VARCHAR类型参数是否为NULL，如果为NULL则返回true，否则返回false。其中：

• @ScalarFunction(value = "is_null", alias = "isnull")声明了函数名为is_null，函数别名为isnull，即在 SQL 中使用is_null和isnull都可以调用该函数
• @Description("Returns TRUE if the argument is NULL")声明了函数描述，使用show functions命令可以看到函数的描述
• @SqlType(StandardTypes.BOOLEAN)声明了函数的返回类型为BOOLEAN
• 因为当函数参数为NULL时，我们不能直接返回NULL，而是要进行判断，所以要加上@SqlNullable避免框架自动返回NULL
• @SqlType(StandardTypes.VARCHAR)声明了函数的参数类型为VARCHAR

注意到，这里使用了 Java 类型Slice来接收 SQL 中VARCHAR类型的值。框架会自动将 SQL 中的数据类型与“原生容器类型”（Native container type）进行绑定，目前“原生容器类型”只包括：boolean、long、double、Slice和Block。VARCHAR对应的原生容器类型是Slice而不是String,Slice的本质是对byte[]进行了封装，为的是更加高效、自由地对内存进行操作。Block可以简单的理解为对应 SQL 中的数组类型。具体的对应关系和绑定过程涉及 Presto 的类型系统和函数调用过程，不是本文讲解的重点，故在此不作展开。

进一步地，我们想对 is_null函数进行升级，使它能够处理任意类型的参数，这时@TypeParameter注解就派上用场了，函数的实现可以改写为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

@ScalarFunction(value = "is_null", alias = "isnull")
@Description("Returns TRUE if the argument is NULL")
public class ExampleIsNullFunction
{
    private IsNullFunctions()
    {
    }

    @TypeParameter("T")
    @SqlType(StandardTypes.BOOLEAN)
    public static boolean isNullSlice(@SqlNullable @SqlType("T") Slice value)
    {
        return (value == null);
    }

    @TypeParameter("T")
    @SqlType(StandardTypes.BOOLEAN)
    public static boolean isNullLong(@SqlNullable @SqlType("T") Long value)
    {
        return (value == null);
    }

    @TypeParameter("T")
    @SqlType(StandardTypes.BOOLEAN)
    public static boolean isNullDouble(@SqlNullable @SqlType("T") Double value)
    {
        return (value == null);
    }

    @TypeParameter("T")
    @SqlType(StandardTypes.BOOLEAN)
    public static boolean isNullBoolean(@SqlNullable @SqlType("T") Boolean value)
    {
        return (value == null);
    }

    @TypeParameter("T")
    @SqlType(StandardTypes.BOOLEAN)
    public static boolean isNullBlock(@SqlNullable @SqlType("T") Block value)
    {
        return (value == null);
    }
}

可以看到，@TypeParameter的使用有点类似 Java 中泛型的用法，类型变量T在声明完之后就可以在@SqlType注解中使用。在实际的调用过程中，框架会将T与实际 SQL 类型进行绑定，然后再去调用以对应的原生容器类型为参数的实际方法。

1.2 聚合函数

聚合的过程一般涉及多行，有一个累积计算的过程，又由于 Presto 是一个分布式的计算引擎，数据分布在多个节点，所以需要用状态对象来维护和记录中间计算结果。

引入状态之后，Presto 将聚合的过程抽象为三个步骤：

1. input(state, value)
2. combine(state1, state2)
3. output(state, out)

首先，input 阶段分别在不同的 worker 中进行，将行值进行累积计算到state中；combine阶段将上一步得到的state进行两两结合；经过前两步，最终会得到一个state，在output阶段对最终的state进行处理输出。

在实现方面，聚合函数的开发使用了和标量函数类似的注解框架，但是由于状态概念的引入，需要定义一个继承于AccumulatorState接口的状态接口，对于简单的聚合，该接口只需要新增聚合所需的getter和setter，框架会自动生成相关的实现和序列化代码；如果聚合过程中需要记录复杂类型（LIST、MAP或自定义的类）的状态，则需要额外实现AccumulatorStateFactory接口和AccumulatorStateSerializer接口，并在状态接口上使用@AccumulatorStateMetadata注解，在注解中指定stateFactoryClass和stateSerializerClass。

下面以实现求DOUBLE类型的列均值的聚合函数avg_double为例来说明如何进行简单聚合函数的开发。

avg_double的聚合状态只需要记录累积和与加数个数，所以状态接口的定义如下：

1
2
3
4
5
6
7
8
9
10
11
12

public interface LongAndDoubleState
        extends AccumulatorState
{
    long getLong();

    void setLong(long value);

    double getDouble();

    void setDouble(double value);
}

使用定义好的状态接口进行聚合函数实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

@AggregationFunction("avg_double")
public class AverageAggregation
{
    @InputFunction
    public static void input(LongAndDoubleState state, @SqlType(StandardTypes.DOUBLE) double value)
    {
        state.setLong(state.getLong() + 1);
        state.setDouble(state.getDouble() + value);
    }

    @CombineFunction
    public static void combine(LongAndDoubleState state, LongAndDoubleState otherState)
    {
        state.setLong(state.getLong() + otherState.getLong());
        state.setDouble(state.getDouble() + otherState.getDouble());
    }

    @OutputFunction(StandardTypes.DOUBLE)
    public static void output(LongAndDoubleState state, BlockBuilder out)
    {
        long count = state.getLong();
        if (count == 0) {
            out.appendNull();
        }
        else {
            double value = state.getDouble();
            DOUBLE.writeDouble(out, value / count);
        }
    }
}

可以看到聚合函数的实现使用了以下注解：

• @AggregationFunction声明了聚合函数的名称，也可以指定函数的别名
• @InputFunction、@CombineFunction和@OutputFunction分别用来标记聚合的三个步骤，其中@OutputFunction注解需要声明聚合函数返回结果的数据类型
• BlockBuilder类为结果输出类，聚合计算出的最终结果值将通过BlockBuilder进行输出

1.3 窗口函数

窗口函数在查询结果的行上进行计算，执行顺序在HAVING子句之后，ORDER BY子句之前。在 Presto SQL 中，窗口函数的语法形式如下：

1

windowFunction(arg1,....argn) OVER([PARTITION BY<...>] [ORDER BY<...>] [RANGE|ROWS BETWEEN AND])

由此可见，窗口函数语法由关键字OVER触发，且包含三个子句：

1. PARTITION BY: 指定输入行分区的规则，类似于聚合函数的GROUP BY子句，不同分区里的计算互不干扰(窗口函数的计算是并发进行的，并发数和partition数量一致)，缺省时将所有数据行视为一个分区
2. ORDER BY: 决定了窗口函数处理输入行的顺序
3. RANGE|ROWS BETWEEN AND: 指定窗口边界，不常用，缺省时的窗口为当前行所在的分区第一行到当前行

窗口函数的开发需要实现WindowFunction接口，WindowFunction接口中声明了两个方法：

• void reset(WindowIndex windowIndex): 处理新分区时，都会调用该方法进行初始化，WindowIndex包含了已排序的分区的所有行
• void processRow(BlockBuilder output, int peerGroupStart, int peerGroupEnd, int frameStart, int frameEnd): 窗口函数的实现方法，BlockBuilder为结果输出类，计算出来的值将通过BlockBuilder进行输出；peerGroupStart和peerGroupEnd为当前处理的行所在的分区的开始和结束的位置；frameStart和frameEnd为当前处理行所在的窗口的开始和结束位置。

实现一个返回窗口中第一个值的窗口函数first_value(x)的代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

@WindowFunctionSignature(name = "first_value", typeVariable = "T", returnType = "T", argumentTypes = "T")
public class FirstValueFunction
        extends WindowFunction
{
    private final int argumentChannel;
    private WindowIndex windowIndex;

    public FirstValueFunction(List<Integer> argumentChannels)
    {
        this.argumentChannel = getOnlyElement(argumentChannels);
    }

    @Override
    public void reset(WindowIndex windowIndex)
    {
        this.windowIndex = windowIndex;
    }

    @Override
    public void processRow(BlockBuilder output, int peerGroupStart, int peerGroupEnd, int frameStart, int frameEnd)
    {
        if (frameStart < 0) {
            output.appendNull();
            return;
        }

        //Outputs a value from the index
        windowIndex.appendTo(argumentChannel, frameStart, output);
    }
}

其中：

• @WindowFunctionSignature注解声明了窗口函数的名称，为了处理任意数据类型的字段，这里还声明了类型变量T，并将返回类型和参数类型都指定为T
• 构造函数中的argumentChannels为参数字段所在列的索引值
• processRow方法中，每次只需要通过列索引argumentChannel和当前行所在的窗口起始索引frameStart，就能确定窗口中的第一个值

2. 函数注册

Presto 函数由MetadataManager中的FunctionRegistry进行管理，开发的函数要生效必须要先注册到FunctionRegistry中。函数注册是在 Presto 服务启动过程中进行的，有以下两种方式进行函数注册。

2.1 内置函数注册

内置函数指的是 Presto 自带的函数库中的函数，函数的实现位于presto-main模块中，在FunctionRegistry初始化时进行注册。具体的注册过程使用了建造者模式，不同类型的函数注册只需要调用FunctionListBuilder对象对应的方法进行注册，关键代码如下：

1
2
3
4
5
6

FunctionListBuilder builder = new FunctionListBuilder()
                .window(RowNumberFunction.class)
                .aggregate(ApproximateCountDistinctAggregation.class)
                .scalar(RepeatFunction.class)
                .function(MAP_HASH_CODE)
                ......

2.2 插件函数注册

内置函数满足不了使用需求时，就需要自行开发函数来拓展函数库。开发者自行编写的拓展函数一般通过插件的方式进行注册。PluginManager在安装插件时会调用插件的getFunctions()方法，将获取到的函数集合通过MetadataManager的addFunctions方法进行注册：

1
2
3
4
5
6
7
8
9

public void installPlugin(Plugin plugin)
    {
        ......
       for (Class<?> functionClass : plugin.getFunctions()) {
            log.info("Registering functions from %s", functionClass.getName());
            metadata.addFunctions(extractFunctions(functionClass));
        }
        ......
    }

所以用做拓展函数库的插件，需要实现getFunctions()方法，来返回拓展的函数集合，例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

public class ExampleFunctionsPlugin
        implements Plugin
{
    @Override
    public Set<Class<?>> getFunctions()
    {
        return ImmutableSet.<Class<?>>builder()
                .add(ExampleNullFunction.class)
                .add(IsNullFunction.class)
                .add(IsEqualOrNullFunction.class)
                .add(ExampleStringFunction.class)
                .add(ExampleAverageFunction.class)
                .build();
    }
}

Trino源码学习-SPI&Plugin

本篇将介绍Trino的SPI和如何通过Plugin体系扩展SPI。Trino 支持通过SPI（Service Provider Interface）方式对其进行扩展点扩展。当前已有的扩展点有:

• Connectors(连接器)
• block encodings(块编码)
• Types(类型)
• Functions(函数)
• System access control(系统访问权限)
• Group provider(资源组)
• Password authenticator(密码验证器)
• Header authenticator(标头验证器)
• Certificate authenticator(证书验证器)
• Event listener(事件侦听器)
• resource group configuration managers(资源组配置管理器)
• session property configuration managers(会话属性配置管理器)
• exchange managers(数据交互管理器)

使用maven插件优化项目

记录可以用于优化项目的maven插件。

蓄水池采样算法

蓄水池采样算法是一种随机抽样算法，它能够在一个很大的集合中，抽取一部分样本，并保证每个样本的选取概率都是相等并随机的。

• 特点：
  • 选取集合可以非常大，甚至不知道边界。
  • 每个样本的选取随机且概率相等。
  • 时间复杂度较低，O(n)，节省内存。
• 适用场景：
  • 在一些非常大的集合，或者未知大小的集合，不知道边界的集合，不知道文件总行数的情况下，随机抽取k个元素。
  • 保证每个元素抽取都是均匀随机的并且概率相等。
  • 尽量高效，节省内存地抽取

Trino源码学习-内存管理

Trino中的内存管理分为两块:

• LocalMemoryManager: 在ServerMainModule中声明，用于管理当前节点的内存使用
• ClusterMemoryManager: 在CoordinatorModule中声明，用于管理集群的内存使用

Trino源码学习-TASK执行

上篇我们介绍了查询是如何调度的。本篇将介绍Task的执行。

HikariCP: Failed to validate connection

最近在日志中常常看到HikariCP的warnning.

1

Failed to validate connection com.mysql.jdbc.JDBC4Connection@xxxxx (...)Possibly consider using a shorter maxLifetime value.

相关配置

HikariCP相关配置

• minimumIdle: 池的最小连接数
• maximumPoolSize: 池的最大连接数
• idleTimeout: 允许连接在池中闲置的最长时间。仅在minimumIdle小于maximumPoolSize时使用。池中连接小于minimumIdle时，空闲连接将不会被取消。
• maxLifetime: 池中连接的最长生命周期。0 表示没有最大生命周期约束。
• keepaliveTime: HikariCP 连接保活的时间间隔。该值必须小于maxLifetime
• connectionTimeout: 控制客户端等待池中连接的最长毫秒数，如果在没有连接可用的情况下超过此时间，则抛出SQLException。

mysql相关配置

• wait_timeout: mysql 为了防止空闲连接浪费，占用资源，在超过wait_timeout 时间后，会主动关闭该连接，清理资源。通过show variables like 'wait_timeout%'可以查询该配置值。

通过对上面配置的了解，结合日志告警，应该是maxLifeTime设置的过长了。导致池中存在已经被close的连接。所以maxLifeTime要小于空闲连接的回收时长(例如 mysql的wait_timeout )。

注意, 应用和数据库中间可能存在HA和DB proxy，maxLifeTime也需要考虑中间链接的空闲超时。

Trino源码学习-查询调度

上篇分析了SQL逻辑执行计划的生成，本篇将继续分析逻辑计划是如何在进行调度的。