如何自定义一个Calcite对Tablesaw查询的适配器

Tablesaw是Java领域中一个比较好用的数据分析工具，可以对数据集进行清洗、转换、统计等。如果能使用Sql对数据集进行查询，那就更完美了。使用Calcite这个开源项目完全可以做到，按照官网所说，Calcite适配了Csv、Elasticsearch、Druid等数据源，适配Tablesaw更不在话下。但是官网中没有提供Tablesaw的适配器，我们可以参考Csv适配器来简单适配Tablesaw。

1、Calcite如何查询Sql

使用Sql查询数据集，想必很多人都会想到Jdbc查询的那一套，其实Calcite就是基于Jdbc来实现的，也是那几个步骤：

/**
 * 1. 加载驱动类
 * 2. 创建Connection
 * 3. 创建Statement
 * 4. 执行查询获取数据集
 * 5. 关闭资源
*/
public static void exec(String sql) throws SQLException {
    Properties info = new Properties();
    Class.forName("org.apache.calcite.jdbc.Driver");    
    try (Connection conn = DriverManager.getConnection("jdbc:calcite:", info);  
        Statement stat = conn.createStatement()) {      
        final ResultSet resultSet = stat.executeQuery(sql);     
        ....
    }
}

Jdbc查询在查询时候会指定连接地址，但是如果对Tablesaw查询要怎么指定连接地址呢？我们可以通过设置查询属性，实现Calcite预留的接口，达到查询Tablesaw的目的。

2、Calcite的连接属性

Calcite的驱动由Avatica提供，可以设置的属性可以查看官网，主要有：

属性	描述
caseSensitive	大小写敏感
lex	语法，默认Oracle语法
quoting	标识符的引用语法：对特殊字段的引用，Oracle："form"
quotedCasing	使用了quoting的标识符，如何排序
unquotedCasing	没有使用quoting的标识符，如何排序
timeZone	默认JVM时区，不建议特别指定
model	模型文件，指定如何处理数据
schema	schema名称
schemaFactory	创建schema的工厂，model存在则不起效
schemaType	schema类型，model存在则不起效
typeSystem	字段类型系统，model存在则不起效

model文件的结构：

{
  "version": "1.0",
  "defaultSchema": "foodmart",
  "schemas": [
    {
      type: 'custom',
      name: 'twissandra',
      factory: 'org.apache.calcite.adapter.cassandra.CassandraSchemaFactory',
      operand: {
        host: 'localhost',
        keyspace: 'twissandra'
      }
    }
  ]
}

schema.json的属性：

属性	描述
version	版本
defaultSchema	默认schema
schemas	schema数组，可有多个schema组成
schemas.name	schema的名称
schemas.type	schema的类型，使用自定义类型需要指定factory
schemas.factory	构造schema的工厂
schemas.operand	构造schema的自定义参数Map
schemas.tables	schema里面可以有多个表
schemas.tables.name	table的名称
schemas.tables.type	table的类型，使用自定义类型需要指定factory
schemas.tables.factory	构造table的工厂
schemas.tables.operand	构造table的自定义参数Map

可以理解为一个schema.json文件里面有多个schema，一个schema里面有多个table。
我们也可以通过url的方式传入参数，效果和上面一样：

jdbc:calcite:schemaFactory=org.apache.calcite.adapter.cassandra.CassandraSchemaFactory; schema.host=localhost; schema.keyspace=twissandra

3、定义Table的结构

我们需要定义Table的结构，来达到结构化查询的目的。

public abstract class DataFrameTable extends AbstractTable {

    protected tech.tablesaw.api.Table table;

    private RelDataType rowType;

    DataFrameTable(tech.tablesaw.api.Table table) {
        this.table = table;
    }

    @Override
    public RelDataType getRowType(RelDataTypeFactory typeFactory) {
        if (rowType == null) {
            rowType = createRelDataType(typeFactory);
        }
        return rowType;
    }

    private RelDataType createRelDataType(RelDataTypeFactory typeFactory) {
        List<RelDataType> types = new ArrayList<>();
        List<String> names = new ArrayList<>();
        for (Column<?> column : table.columns()) {
            DataFrameFieldType type = DataFrameFieldType.of(column.type());
            RelDataType relDataType = type.toType((JavaTypeFactory) typeFactory);
            types.add(relDataType);
            names.add(column.name());
        }
        return typeFactory.createStructType(Pair.zip(names, types));
    }
}

定义Table，必不可少定义字段数据类型，需要指定Tablesaw的字段类型和Calcite的字段类型的对应关系

展开查看

enum DataFrameFieldType {
    STRING(String.class, ColumnType.STRING),
    TEXT(String.class, ColumnType.TEXT),
    BOOLEAN(Primitive.BOOLEAN, ColumnType.BOOLEAN),
    SHORT(Primitive.SHORT, ColumnType.SHORT),
    INT(Primitive.INT, ColumnType.INTEGER),
    LONG(Primitive.LONG, ColumnType.LONG),
    FLOAT(Primitive.FLOAT, ColumnType.FLOAT),
    DOUBLE(Primitive.DOUBLE, ColumnType.DOUBLE),
    DATE(java.sql.Date.class, ColumnType.LOCAL_DATE),
    TIME(java.sql.Time.class, ColumnType.LOCAL_TIME),
    TIMESTAMP(java.sql.Timestamp.class, ColumnType.LOCAL_DATE_TIME);

    private final Class<?> clazz;
    private final ColumnType columnType;

    private static final Map<ColumnType, DataFrameFieldType> MAP = new HashMap<>();

    static {
        for (DataFrameFieldType value : values()) {
            MAP.put(value.columnType, value);
        }
    }

    DataFrameFieldType(Primitive primitive, ColumnType columnType) {
        this(primitive.boxClass, columnType);
    }

    DataFrameFieldType(Class<?> clazz, ColumnType columnType) {
        this.clazz = clazz;
        this.columnType = columnType;
    }

    public RelDataType toType(JavaTypeFactory typeFactory) {
        RelDataType javaType = typeFactory.createJavaType(clazz);
        RelDataType sqlType = typeFactory.createSqlType(javaType.getSqlTypeName());
        return typeFactory.createTypeWithNullability(sqlType, true);
    }

    public static DataFrameFieldType of(ColumnType columnType) {
        return MAP.get(columnType);
    }
}

这样我们就定义好一个Table的结构了，我们定义的Table中，有一个tech.tablesaw.api.Table，这个是用于提供数据和字段结构的，至于怎么来的，后面细说。

4、定义Table的数据枚举器Enumerator

Calcite定义了3中查询Table的方式，可以参考

• ScannableTable：根据全部数据查询
• FilterableTable：查询底层DB时进行一部分的数据过滤，再在内存中查询
• TranslatableTable：自定义优化规则

我们以最简单的ScannableTable为例子查询。

public class DataFrameScannableTable extends DataFrameTable implements ScannableTable {

    DataFrameScannableTable(tech.tablesaw.api.Table table) {
        super(table);
    }

    public Enumerable<Object[]> scan(DataContext root) {
        return new AbstractEnumerable<Object[]>() {
            public Enumerator<Object[]> enumerator() {
                return new DataFrameEnumerator(table);
            }
        };
    }
}

ScannableTable的scan方法定义了如何获取Enumerator，而Enumerator是操作数据的关键

展开查看

class DataFrameEnumerator implements Enumerator<Object[]> {

    private Enumerator<Object[]> enumerator;

    DataFrameEnumerator(tech.tablesaw.api.Table table) {
        List<Object[]> objs = new ArrayList<>();
        for (int row = 0; row < table.rowCount(); row++) {
            Object[] rows = new Object[table.columnCount()];
            for (int col = 0; col < table.columnCount(); col++) {
                Column<?> column = table.column(col);
                rows[col] = convertToEnumeratorObject(column, row);
            }
            objs.add(rows);
        }
        this.enumerator = Linq4j.enumerator(objs);
    }

    public Object[] current() {
        return enumerator.current();
    }

    public boolean moveNext() {
        return enumerator.moveNext();
    }

    public void reset() {
        enumerator.reset();
    }

    public void close() {
        enumerator.close();
    }

    private Object convertToEnumeratorObject(Column<?> column, int row) {
        final TimeZone gmt = TimeZone.getTimeZone("GMT");
        if (column instanceof DateColumn) {
            return ((DateColumn) column).get(row).toEpochDay();
        } else if (column instanceof TimeColumn) {
            return Date.from(
                    ((TimeColumn) column).get(row)
                            .atDate(LocalDate.ofEpochDay(0))
                            .atZone(gmt.toZoneId())
                            .toInstant()
            ).getTime();
        } else if (column instanceof DateTimeColumn) {
            return Date.from(
                    ((DateTimeColumn) column).get(row)
                            .atZone(gmt.toZoneId())
                            .toInstant()
            ).getTime();
        } else {
            return column.get(row);
        }
    }
}

5、使用其他方法代替model.json传入参数

至此，我们只需让Calcite使用我们定义的Table来查询数据就行了，前面说到可以通过传入model.json来实现。我们通过自定义的TableFactory来创建Table：

public class DataFrameTableFactory implements TableFactory<DataFrameTable> {

    private tech.tablesaw.api.Table table;

    public DataFrameTableFactory(tech.tablesaw.api.Table table) {
        this.table = table;
    }

    @Override
    public DataFrameTable create(SchemaPlus schema,
                                 String name,
                                 Map<String, Object> operand,
                                 RelDataType rowType) {
        return new DataFrameScannableTable(table);
    }
}

但是我们看到tech.tablesaw.api.Table参数，这个该怎么传入？如果在有数据的前提下，我们很难通过model.json的operand参数来传入，model.json适合根据operand参数到数据源获取数据后再操作。对此，我们使用另外一个方法：

    private static void setTableModel(Connection connection, tech.tablesaw.api.Table table) {
        SchemaPlus rootSchema = ((CalciteConnection) connection).getRootSchema();
        TableFactory<?> tableFactory = new DataFrameTableFactory(table);
        org.apache.calcite.schema.Table t = tableFactory
                .create(rootSchema, table.name(), ImmutableMap.of(), null);
        rootSchema.add(table.name(), t);
    }

该方法主要是获取CalciteConnection的SchemaPlus，来传入我们定义的Table。

6、测试

到此，我们应该能用Sql查询Tablesaw了。我们来测试一下：

展开查看

    @Test
    public void test() throws SQLException {
        tech.tablesaw.api.Table table = tech.tablesaw.api.Table.create("test");

        StringColumn stringColumn = StringColumn.create("A");
        stringColumn.append("bbbbb");
        table.addColumns(stringColumn);

        DateColumn dateColumn = DateColumn.create("B");
        dateColumn.append(LocalDate.now());
        table.addColumns(dateColumn);

        DateTimeColumn dateTimeColumn = DateTimeColumn.create("C");
        dateTimeColumn.append(LocalDateTime.now());
        table.addColumns(dateTimeColumn);

        TimeColumn timeColumn = TimeColumn.create("D");
        timeColumn.append(LocalTime.now());
        table.addColumns(timeColumn);

        Table t = DataFrameQueryUtils.exec(table, "SELECT * FROM "test"");
        System.out.println(t);
    }

使用各种Calcite内置的函数也是没问题的，但是如果使用了额外的函数，可能需要额外定义函数的计算方式。

7、总结

这个例子只用了Table，并没有使用Schema，其实Schema的原理也差不多，就是定义Table的集合，有需要可以参考官方来自己实现。Calcite做Tablesaw的适配器也不在话下，用二维数组代替Tablesaw也是可以的。虽然如此，但是还是要注意关于时间类型的一些坑，是关于时间类型转换和时区的一些问题，这个有空再总结。