Presto コネクターを実装する第二回

こんにちは。松本です。

前回に続き、Presto のコネクター実装方法について解説します。第二回の今回はテーブルのハンドルクラスやメタデータを操作するクラスを扱います。

実装する主なクラスは次の 4 つです。

TechscoreColumnHandle implements com.facebook.presto.spi.ConnectorColumnHandle
TechscoreTableHandle implements com.facebook.presto.spi.ConnectorTableHandle
TechscoreHandleResolver implements com.facebook.presto.spi.ConnectorHandleResolver
TechscoreMetadata extends com.facebook.presto.spi.ReadOnlyConnectorMetadata

ハンドルクラス

Presto コネクターでは、ConnectorTableHandle インターフェースと ConnectorColumnHandle インターフェースの実装クラスを使って、テーブルやカラムをハンドリングします。

まずはカラムハンドル。

com.techscore.example.presto.plugin.TechscoreColumnHandle

package com.techscore.example.presto.plugin;

import lombok.EqualsAndHashCode;
import lombok.NonNull;
import lombok.ToString;

import com.facebook.presto.spi.ColumnMetadata;
import com.facebook.presto.spi.ConnectorColumnHandle;
import com.facebook.presto.spi.type.Type;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.primitives.Ints;

@EqualsAndHashCode
@ToString
public class TechscoreColumnHandle implements ConnectorColumnHandle, Comparable<TechscoreColumnHandle> {

    private final String connectorId;
    private final String caseSensitiveName;
    private final String name;
    private final Type type;
    private final int ordinalPosition;

    @JsonCreator
    public TechscoreColumnHandle(
            @JsonProperty("connectorId") @NonNull String connectorId,
            @JsonProperty("caseSensitiveName") @NonNull String caseSensitiveName,
            @JsonProperty("type") @NonNull Type type,
            @JsonProperty("ordinalPosition") int ordinalPosition) {
        this.connectorId = connectorId;
        this.caseSensitiveName = caseSensitiveName;
        this.name = caseSensitiveName.toLowerCase();
        this.type = type;
        this.ordinalPosition = ordinalPosition;
    }

    @JsonProperty
    public String getConnectorId() {
        return connectorId;
    }

    @JsonProperty
    public String getCaseSensitiveName() {
        return caseSensitiveName;
    }

    @JsonIgnore
    public String getName() {
        return name;
    }

    @JsonProperty
    public Type getType() {
        return type;
    }

    @JsonProperty
    public int getOrdinalPosition() {
        return ordinalPosition;
    }

    public ColumnMetadata toColumnMetadata() {
        return new ColumnMetadata(getName(), getType(), getOrdinalPosition(), false);
    }

    @Override
    public int compareTo(TechscoreColumnHandle otherHandle) {
        return Ints.compare(getOrdinalPosition(), otherHandle.getOrdinalPosition());
    }
}

package com.techscore.example.presto.plugin;

import lombok.EqualsAndHashCode;

import lombok.NonNull;

import lombok.ToString;

import com.facebook.presto.spi.ColumnMetadata;

import com.facebook.presto.spi.ConnectorColumnHandle;

import com.facebook.presto.spi.type.Type;

import com.fasterxml.jackson.annotation.JsonCreator;

import com.fasterxml.jackson.annotation.JsonIgnore;

import com.fasterxml.jackson.annotation.JsonProperty;

import com.google.common.primitives.Ints;

@EqualsAndHashCode

@ToString

public class TechscoreColumnHandle implements ConnectorColumnHandle, Comparable<TechscoreColumnHandle> {

private final String connectorId;

private final String caseSensitiveName;

private final String name;

private final Type type;

private final int ordinalPosition;

@JsonCreator

public TechscoreColumnHandle(

@JsonProperty("connectorId") @NonNull String connectorId,

@JsonProperty("caseSensitiveName") @NonNull String caseSensitiveName,

@JsonProperty("type") @NonNull Type type,

@JsonProperty("ordinalPosition") int ordinalPosition) {

this.connectorId = connectorId;

this.caseSensitiveName = caseSensitiveName;

this.name = caseSensitiveName.toLowerCase();

this.type = type;

this.ordinalPosition = ordinalPosition;

}

@JsonProperty

public String getConnectorId() {

return connectorId;

}

@JsonProperty

public String getCaseSensitiveName() {

return caseSensitiveName;

}

@JsonIgnore

public String getName() {

return name;

}

@JsonProperty

public Type getType() {

return type;

}

@JsonProperty

public int getOrdinalPosition() {

return ordinalPosition;

}

public ColumnMetadata toColumnMetadata() {

return new ColumnMetadata(getName(), getType(), getOrdinalPosition(), false);

}

@Override

public int compareTo(TechscoreColumnHandle otherHandle) {

return Ints.compare(getOrdinalPosition(), otherHandle.getOrdinalPosition());

}

ConnectorColumnHandle 実装クラスは、コネクター内でカラムを扱いやすいよう好きに実装します。

実装のポイントとして、Presto 本体側で本クラスのインスタンスがキャッシュされるよう、@JsonCreator や @JsonProperty アノテーションを使って JSON でのシリアライズ／デシリアライズを可能とすることです。

Presto ではケース無視でカラム名が扱われる為、caseSensitiveName で渡されたカラム名を小文字化した name プロパティも持つようにしています。

        this.name = caseSensitiveName.toLowerCase();

1	this.name = caseSensitiveName.toLowerCase();

type プロパティはカラムのデータ型を表すクラスで、com.facebook.presto.spi.type パッケージ内に基本的なデータ型が用意されています。

主なデータ型です。

ordinalPosition プロパティはテーブル内でのこのカラムの順序を表します。このプロパティは Presto が使用するカラムメタデータクラスである ColumnMetadata にも引き継がれ、SELECT * 等で SQL が発行された場合のカラムの並び順として利用されます。

最終的に、これらの情報を使い ColumnMetadata を生成します。

次にテーブルハンドル。

com.techscore.example.presto.plugin.TechscoreTableHandle

package com.techscore.example.presto.plugin;

import lombok.EqualsAndHashCode;
import lombok.NonNull;
import lombok.ToString;

import com.facebook.presto.spi.ConnectorTableHandle;
import com.facebook.presto.spi.SchemaTableName;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;

@EqualsAndHashCode
@ToString
public class TechscoreTableHandle implements ConnectorTableHandle {

    private final String connectorId;
    private final SchemaTableName schemaTableName;

    @JsonCreator
    public TechscoreTableHandle(
            @JsonProperty("connectorId") @NonNull String connectorId,
            @JsonProperty("schemaTableName") @NonNull SchemaTableName schemaTableName) {
        this.connectorId = connectorId;
        this.schemaTableName = schemaTableName;
    }

    @JsonProperty
    public String getConnectorId() {
        return connectorId;
    }

    @JsonProperty
    public SchemaTableName getSchemaTableName() {
        return schemaTableName;
    }

    @JsonIgnore
    public String getSchemaName() {
        return getSchemaTableName().getSchemaName();
    }

    @JsonIgnore
    public String getTableName() {
        return getSchemaTableName().getTableName();
    }
}

package com.techscore.example.presto.plugin;

import lombok.EqualsAndHashCode;

import lombok.NonNull;

import lombok.ToString;

import com.facebook.presto.spi.ConnectorTableHandle;

import com.facebook.presto.spi.SchemaTableName;

import com.fasterxml.jackson.annotation.JsonCreator;

import com.fasterxml.jackson.annotation.JsonIgnore;

import com.fasterxml.jackson.annotation.JsonProperty;

@EqualsAndHashCode

@ToString

public class TechscoreTableHandle implements ConnectorTableHandle {

private final String connectorId;

private final SchemaTableName schemaTableName;

@JsonCreator

public TechscoreTableHandle(

@JsonProperty("connectorId") @NonNull String connectorId,

@JsonProperty("schemaTableName") @NonNull SchemaTableName schemaTableName) {

this.connectorId = connectorId;

this.schemaTableName = schemaTableName;

}

@JsonProperty

public String getConnectorId() {

return connectorId;

}

@JsonProperty

public SchemaTableName getSchemaTableName() {

return schemaTableName;

}

@JsonIgnore

public String getSchemaName() {

return getSchemaTableName().getSchemaName();

}

@JsonIgnore

public String getTableName() {

return getSchemaTableName().getTableName();

}

ConnectorTableHandle の実装クラスも、コネクター内でテーブル情報が扱いやすいよう好きに実装します。

今回の実装では connectorId と SchemaTableName のインスタンスを持つだけとしています。SchemaTableName はその名の通り対象テーブルのスキーマ名とテーブル名を持つクラスです。

こちらも TechscoreColumnHandle 同様、JSON でのシリアライズ／デシリアライズが可能になるようアノテーションを付けています。TechscoreTableHandle では JSON でのキャッシュ化を見込んで、あまりファットなオブジェクトにならないよう考慮しています。

com.techscore.example.presto.plugin.TechscoreHandleResolver

package com.techscore.example.presto.plugin;

import static com.google.common.base.Preconditions.checkArgument;

import javax.inject.Inject;

import lombok.NonNull;
import lombok.Value;

import com.facebook.presto.spi.ConnectorColumnHandle;
import com.facebook.presto.spi.ConnectorHandleResolver;
import com.facebook.presto.spi.ConnectorIndexHandle;
import com.facebook.presto.spi.ConnectorInsertTableHandle;
import com.facebook.presto.spi.ConnectorOutputTableHandle;
import com.facebook.presto.spi.ConnectorSplit;
import com.facebook.presto.spi.ConnectorTableHandle;
import com.google.inject.name.Named;

@Value
public class TechscoreHandleResolver implements ConnectorHandleResolver {

    private final String connectorId;

    @Inject
    TechscoreHandleResolver(@Named("connectorId") @NonNull String connectorId) {
        this.connectorId = connectorId;
    }

    @Override
    public boolean canHandle(ConnectorTableHandle tableHandle) {
        return tableHandle != null && tableHandle instanceof TechscoreTableHandle
                && getConnectorId().equals(((TechscoreTableHandle) tableHandle).getConnectorId());
    }

    @Override
    public boolean canHandle(ConnectorColumnHandle columnHandle) {
        return columnHandle != null && columnHandle instanceof TechscoreColumnHandle
                && getConnectorId().equals(((TechscoreColumnHandle) columnHandle).getConnectorId());
    }

    @Override
    public boolean canHandle(ConnectorSplit split) {
        return split != null && split instanceof TechscoreSplit
                && getConnectorId().equals(((TechscoreSplit) split).getConnectorId());
    }

    @Override
    public boolean canHandle(ConnectorIndexHandle indexHandle) {
        return false;
    }

    @Override
    public boolean canHandle(ConnectorOutputTableHandle tableHandle) {
        return false;
    }

    @Override
    public boolean canHandle(ConnectorInsertTableHandle tableHandle) {
        return false;
    }

    @Override
    public Class<? extends ConnectorTableHandle> getTableHandleClass() {
        return TechscoreTableHandle.class;
    }

    @Override
    public Class<? extends ConnectorColumnHandle> getColumnHandleClass() {
        return TechscoreColumnHandle.class;
    }

    @Override
    public Class<? extends ConnectorSplit> getSplitClass() {
        return TechscoreSplit.class;
    }

    @Override
    public Class<? extends ConnectorIndexHandle> getIndexHandleClass() {
        throw new UnsupportedOperationException();
    }

    @Override
    public Class<? extends ConnectorOutputTableHandle> getOutputTableHandleClass() {
        throw new UnsupportedOperationException();
    }

    @Override
    public Class<? extends ConnectorInsertTableHandle> getInsertTableHandleClass() {
        throw new UnsupportedOperationException();
    }

    TechscoreTableHandle convertTableHandle(ConnectorTableHandle tableHandle) {
        checkArgument(canHandle(tableHandle));
        return (TechscoreTableHandle) tableHandle;
    }

    TechscoreColumnHandle convertColumnHandle(ConnectorColumnHandle columnHandle) {
        checkArgument(canHandle(columnHandle));
        return (TechscoreColumnHandle) columnHandle;
    }

    TechscoreSplit convertSplit(ConnectorSplit split) {
        checkArgument(canHandle(split));
        return (TechscoreSplit) split;
    }
}

100

101

102

103

104

105

106

package com.techscore.example.presto.plugin;

import static com.google.common.base.Preconditions.checkArgument;

import javax.inject.Inject;

import lombok.NonNull;

import lombok.Value;

import com.facebook.presto.spi.ConnectorColumnHandle;

import com.facebook.presto.spi.ConnectorHandleResolver;

import com.facebook.presto.spi.ConnectorIndexHandle;

import com.facebook.presto.spi.ConnectorInsertTableHandle;

import com.facebook.presto.spi.ConnectorOutputTableHandle;

import com.facebook.presto.spi.ConnectorSplit;

import com.facebook.presto.spi.ConnectorTableHandle;

import com.google.inject.name.Named;

@Value

public class TechscoreHandleResolver implements ConnectorHandleResolver {

private final String connectorId;

@Inject

TechscoreHandleResolver(@Named("connectorId") @NonNull String connectorId) {

this.connectorId = connectorId;

}

@Override

public boolean canHandle(ConnectorTableHandle tableHandle) {

return tableHandle != null && tableHandle instanceof TechscoreTableHandle

&& getConnectorId().equals(((TechscoreTableHandle) tableHandle).getConnectorId());

}

@Override

public boolean canHandle(ConnectorColumnHandle columnHandle) {

return columnHandle != null && columnHandle instanceof TechscoreColumnHandle

&& getConnectorId().equals(((TechscoreColumnHandle) columnHandle).getConnectorId());

}

@Override

public boolean canHandle(ConnectorSplit split) {

return split != null && split instanceof TechscoreSplit

&& getConnectorId().equals(((TechscoreSplit) split).getConnectorId());

}

@Override

public boolean canHandle(ConnectorIndexHandle indexHandle) {

return false;

}

@Override

public boolean canHandle(ConnectorOutputTableHandle tableHandle) {

return false;

}

@Override

public boolean canHandle(ConnectorInsertTableHandle tableHandle) {

return false;

}

@Override

public Class<? extends ConnectorTableHandle> getTableHandleClass() {

return TechscoreTableHandle.class;

}

@Override

public Class<? extends ConnectorColumnHandle> getColumnHandleClass() {

return TechscoreColumnHandle.class;

}

@Override

public Class<? extends ConnectorSplit> getSplitClass() {

return TechscoreSplit.class;

}

@Override

public Class<? extends ConnectorIndexHandle> getIndexHandleClass() {

throw new UnsupportedOperationException();

}

@Override

public Class<? extends ConnectorOutputTableHandle> getOutputTableHandleClass() {

throw new UnsupportedOperationException();

}

@Override

public Class<? extends ConnectorInsertTableHandle> getInsertTableHandleClass() {

throw new UnsupportedOperationException();

}

TechscoreTableHandle convertTableHandle(ConnectorTableHandle tableHandle) {

checkArgument(canHandle(tableHandle));

return (TechscoreTableHandle) tableHandle;

}

TechscoreColumnHandle convertColumnHandle(ConnectorColumnHandle columnHandle) {

checkArgument(canHandle(columnHandle));

return (TechscoreColumnHandle) columnHandle;

}

TechscoreSplit convertSplit(ConnectorSplit split) {

checkArgument(canHandle(split));

return (TechscoreSplit) split;

}

ConnectorHandleResulver インターフェースの実装クラスはこんなものと見て頂けば十分です。各種ハンドルクラスを扱うシンプルなメソッドを持っているだけのクラスです。

今回の実装では ConnectorTableHandle, ConnectorColumnHandle, ConnectorSplit 以外は扱わないので、これらに関係しないメソッドは UnsupportedOperationException をスローしたり、false を返すのみとしました。

canHandle() メソッドや getSplitClass() 等で使われている ConnectorSplit インターフェースとその実装クラスである TechscoreSplit については次回、触れる予定です。

メタデータ操作クラス

次は重要なメタデータクラス。Presto は ConnectorMetadata インターフェースの実装クラスを通して、コネクターが扱う各種メタデータにアクセスします。

com.techscore.example.presto.plugin.TechscoreMetadata

package com.techscore.example.presto.plugin;

import static java.util.stream.Collectors.toList;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import javax.inject.Inject;

import lombok.Getter;
import lombok.NonNull;
import lombok.val;

import org.weakref.jmx.internal.guava.collect.ImmutableList;

import com.facebook.presto.spi.ColumnMetadata;
import com.facebook.presto.spi.ConnectorColumnHandle;
import com.facebook.presto.spi.ConnectorSession;
import com.facebook.presto.spi.ConnectorTableHandle;
import com.facebook.presto.spi.ConnectorTableMetadata;
import com.facebook.presto.spi.ReadOnlyConnectorMetadata;
import com.facebook.presto.spi.SchemaTableName;
import com.facebook.presto.spi.SchemaTablePrefix;
import com.google.common.collect.ImmutableMap;
import com.google.inject.name.Named;

public class TechscoreMetadata extends ReadOnlyConnectorMetadata {

    @Getter
    private final String connectorId;

    @Getter
    private final TechscoreClient tClient;

    @Getter
    private final TechscoreHandleResolver tHandleResolver;

    @Getter
    private final TechscoreConnectorConfig tConnectorConfig;

    @Inject
    public TechscoreMetadata(@Named("connectorId") @NonNull String connectorId,
            @NonNull TechscoreClient tClient,
            @NonNull TechscoreHandleResolver tHandleResolver,
            @NonNull TechscoreConnectorConfig tConnectorConfig) {
        this.connectorId = connectorId;
        this.tClient = tClient;
        this.tHandleResolver = tHandleResolver;
        this.tConnectorConfig = tConnectorConfig;
    }

    @Override
    public List<String> listSchemaNames(ConnectorSession session) {
        return getTClient().listSchemaNames();
    }

    @Override
    public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName) {
        return getTClient().getTTableHandle(schemaTableName);
    }

    @Override
    public ConnectorTableMetadata getTableMetadata(ConnectorTableHandle tableHandle) {
        val tTableHandle = getTHandleResolver().convertTableHandle(tableHandle);
        if (tTableHandle == null) {
            return null;
        }

        val schemaTableName = tTableHandle.getSchemaTableName();
        val tColumnHandles = getTClient().getTColumnHandles(schemaTableName);
        val columnMetadatas = new ArrayList<ColumnMetadata>();
        tColumnHandles.forEach(t -> columnMetadatas.add(t.toColumnMetadata()));

        return new ConnectorTableMetadata(schemaTableName, columnMetadatas);
    }

    @Override
    public Map<String, ConnectorColumnHandle> getColumnHandles(ConnectorTableHandle tableHandle) {
        val tTableHandle = getTHandleResolver().convertTableHandle(tableHandle);
        val tColumnHandles = getTClient().getTColumnHandles(tTableHandle.getSchemaTableName());
        val columnHandleMap = new HashMap<String, ConnectorColumnHandle>();
        tColumnHandles.forEach(t -> columnHandleMap.put(t.getName(), t));
        return columnHandleMap;
    }

    @Override
    public ColumnMetadata getColumnMetadata(ConnectorTableHandle tableHandle, ConnectorColumnHandle columnHandle) {
        getTHandleResolver().convertTableHandle(tableHandle);
        val tColumnHandle = getTHandleResolver().convertColumnHandle(columnHandle);
        return tColumnHandle.toColumnMetadata();
    }

    @Override
    public List<SchemaTableName> listTables(ConnectorSession session, String schemaNameOrNull) {
        return getTClient().listSchemaTableNames().stream()
                .filter(t -> schemaNameOrNull == null || schemaNameOrNull.equals(t.getSchemaName()))
                .collect(toList());
    }

    @Override
    public ConnectorColumnHandle getSampleWeightColumnHandle(ConnectorTableHandle tableHandle) {
        return null;
    }

    @Override
    public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(
            ConnectorSession session, @NonNull SchemaTablePrefix prefix) {
        val tableColumns = ImmutableMap.<SchemaTableName, List<ColumnMetadata>>builder();
        for (val schemaTableName : getTClient().listSchemaTableNames()) {
            if (prefix.matches(schemaTableName)) {
                val columnMetadataList = getTClient().getTColumnHandles(schemaTableName)
                        .stream().map(TechscoreColumnHandle::toColumnMetadata)
                        .collect(toList());
                tableColumns.put(schemaTableName, ImmutableList.copyOf(columnMetadataList));
            }
        }
        return tableColumns.build();
    }

}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

package com.techscore.example.presto.plugin;

import static java.util.stream.Collectors.toList;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.stream.Collectors;

import javax.inject.Inject;

import lombok.Getter;

import lombok.NonNull;

import lombok.val;

import org.weakref.jmx.internal.guava.collect.ImmutableList;

import com.facebook.presto.spi.ColumnMetadata;

import com.facebook.presto.spi.ConnectorColumnHandle;

import com.facebook.presto.spi.ConnectorSession;

import com.facebook.presto.spi.ConnectorTableHandle;

import com.facebook.presto.spi.ConnectorTableMetadata;

import com.facebook.presto.spi.ReadOnlyConnectorMetadata;

import com.facebook.presto.spi.SchemaTableName;

import com.facebook.presto.spi.SchemaTablePrefix;

import com.google.common.collect.ImmutableMap;

import com.google.inject.name.Named;

public class TechscoreMetadata extends ReadOnlyConnectorMetadata {

@Getter

private final String connectorId;

@Getter

private final TechscoreClient tClient;

@Getter

private final TechscoreHandleResolver tHandleResolver;

@Getter

private final TechscoreConnectorConfig tConnectorConfig;

@Inject

public TechscoreMetadata(@Named("connectorId") @NonNull String connectorId,

@NonNull TechscoreClient tClient,

@NonNull TechscoreHandleResolver tHandleResolver,

@NonNull TechscoreConnectorConfig tConnectorConfig) {

this.connectorId = connectorId;

this.tClient = tClient;

this.tHandleResolver = tHandleResolver;

this.tConnectorConfig = tConnectorConfig;

}

@Override

public List<String> listSchemaNames(ConnectorSession session) {

return getTClient().listSchemaNames();

}

@Override

public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName) {

return getTClient().getTTableHandle(schemaTableName);

}

@Override

public ConnectorTableMetadata getTableMetadata(ConnectorTableHandle tableHandle) {

val tTableHandle = getTHandleResolver().convertTableHandle(tableHandle);

if (tTableHandle == null) {

return null;

}

val schemaTableName = tTableHandle.getSchemaTableName();

val tColumnHandles = getTClient().getTColumnHandles(schemaTableName);

val columnMetadatas = new ArrayList<ColumnMetadata>();

tColumnHandles.forEach(t -> columnMetadatas.add(t.toColumnMetadata()));

return new ConnectorTableMetadata(schemaTableName, columnMetadatas);

}

@Override

public Map<String, ConnectorColumnHandle> getColumnHandles(ConnectorTableHandle tableHandle) {

val tTableHandle = getTHandleResolver().convertTableHandle(tableHandle);

val tColumnHandles = getTClient().getTColumnHandles(tTableHandle.getSchemaTableName());

val columnHandleMap = new HashMap<String, ConnectorColumnHandle>();

tColumnHandles.forEach(t -> columnHandleMap.put(t.getName(), t));

return columnHandleMap;

}

@Override

public ColumnMetadata getColumnMetadata(ConnectorTableHandle tableHandle, ConnectorColumnHandle columnHandle) {

getTHandleResolver().convertTableHandle(tableHandle);

val tColumnHandle = getTHandleResolver().convertColumnHandle(columnHandle);

return tColumnHandle.toColumnMetadata();

}

@Override

public List<SchemaTableName> listTables(ConnectorSession session, String schemaNameOrNull) {

return getTClient().listSchemaTableNames().stream()

.filter(t -> schemaNameOrNull == null || schemaNameOrNull.equals(t.getSchemaName()))

.collect(toList());

}

@Override

public ConnectorColumnHandle getSampleWeightColumnHandle(ConnectorTableHandle tableHandle) {

return null;

}

@Override

public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(

ConnectorSession session, @NonNull SchemaTablePrefix prefix) {

val tableColumns = ImmutableMap.<SchemaTableName, List<ColumnMetadata>>builder();

for (val schemaTableName : getTClient().listSchemaTableNames()) {

if (prefix.matches(schemaTableName)) {

val columnMetadataList = getTClient().getTColumnHandles(schemaTableName)

.stream().map(TechscoreColumnHandle::toColumnMetadata)

.collect(toList());

tableColumns.put(schemaTableName, ImmutableList.copyOf(columnMetadataList));

}

return tableColumns.build();

}

ConnectorMetadata は、ConnectorTableHandle インターフェースおよび TechscoreColumnHandle インターフェースを操作する重要なインターフェースです。その実装クラスである TechscoreMetadata は、ConnectorMetadata の抽象クラスである ReadOnlyConnectorMetadata のサブクラスとして実装しています。

コンストラクタで注入している TechscoreClient は本サンプルコネクターの為に用意した POJO で、仮想の「TECHSCORE ブログ API クライアント」です。サンプルコネクターはこのクラスを通してテーブルのメタデータやレコードを取得します（今回はサンプルなので静的に情報を返しているだけのクラスとなっています）。TechscoreClient とその関連クラスについては本記事の最後にソースコードを掲載します。

listSchemaNames() メソッドはコネクターが扱うスキーマの名前を返すもので、show schemas from techscore; 等のコマンドを実行した場合に利用されます。

getTableHandle() メソッドや getColumnHandles() メソッドは指定されたテーブルのテーブルハンドルやカラムハンドルを返すものです。シグニチャ的には戻り値の型が ConnectorTableHandle や ConnectorColumnHandle となっていますが、本コネクターでの実装クラスである TechscoreTableHandle, TechscoreColumnHandle を返すよう実装します。

getTableMetadata() メソッドや getColumnMetadata() メソッドは、ハンドルを ConnectorTableMetadata や ColumnMetadata といったメタデータに変換するものです。引数は型として ConnectorTableHandle インターフェースや ConnectorColumnHandle インターフェースが定義されていますが、実際にはその実装クラスである TechscoreTableHandle, TechscoreColumnHandle のインスタンスが渡されます。

listTables() メソッドはコネクターが扱うテーブルの SchemaTableName オブジェクトのリストを返すもので、show tables from techscore.schema1; のようなコマンド発行時に利用されます。ここで返す SchemaTableName のリストは、引数 String schemaNameOrNull が指定するスキーマに属すテーブルについてだけを扱いますが、schemaNameOrNull の値が null の場合はスキーマに関係なく全てのテーブルの SchemaTableName を返すよう実装します。

listTableColumns() メソッドは引数に従って対象テーブルのカラムリストを返します。引数の SchemaTablePrefix は SchemaTableName と同様、スキーマ名とテーブル名を持つクラスですが、SchemaTableName とは違い各プロパティが null を許可していますので、その内容によって次のように処理を分ける必要があります。

「スキーマ名もテーブル名も null 」 - 全テーブルを対象とし、テーブルごとのカラムリストを作成する
「テーブル名のみ null 」 - 指定されたスキーマに属す全てのテーブルを対象とし、テーブルごとのカラムリストを作成する
「いずれも null ではない」 - 指定されたひとつのテーブルを対象とし、カラムリストを作成する

ここで役に立つのが SchemaTablePrefix#matches(SchemaTableName) メソッドで、引数として渡した SchemaTableName オブジェクトが SchemaTablePrefix にマッチするかを上手く判断してくれます。

テーブル定義とデータ

最後に TechscoreMetadata で少し触れた TechscoreClient とその関連クラスのソースを掲載します。

「記事テーブル / techscore.schema1.entries 」と「著者テーブル / techscore.schema1.authors 」を定義しています。

com.techscore.example.presto.plugin.TechscoreClient

package com.techscore.example.presto.plugin;

import static java.util.stream.Collectors.toList;

import java.util.List;
import java.util.Map;

import javax.inject.Inject;

import lombok.Getter;
import lombok.NonNull;
import lombok.val;

import org.weakref.jmx.internal.guava.collect.Maps;

import com.facebook.presto.spi.SchemaTableName;
import com.google.inject.name.Named;

public class TechscoreClient {

    @Getter
    private final String connectorId;

    @Getter
    private final TechscoreConnectorConfig tConnectorConfig;

    @Getter
    private final Map<SchemaTableName, TechscoreTable> tTableMap;

    @Inject
    public TechscoreClient(@Named("connectorId") @NonNull String connectorId,
            @NonNull TechscoreConnectorConfig tConnectorConfig) {
        this.connectorId = connectorId;
        this.tConnectorConfig = tConnectorConfig;
        this.tTableMap = Maps.uniqueIndex(TechscoreTables.list(connectorId, tConnectorConfig.getUrlBase()), t -> t.getSchemaTableName());
    }

    public List<String> listSchemaNames() {
        return getTTableMap().values().stream().map(TechscoreTable::getSchemaName).collect(toList());
    }

    public List<SchemaTableName> listSchemaTableNames() {
        return getTTableMap().values().stream().map(TechscoreTable::getSchemaTableName).collect(toList());
    }

    public TechscoreTableHandle getTTableHandle(@NonNull SchemaTableName schemaTableName) {
        val table = getTTableMap().get(schemaTableName);
        if (table == null) {
            return null;
        }
        return table.getTTableHandle();
    }

    public List<TechscoreColumnHandle> getTColumnHandles(@NonNull SchemaTableName schemaTableName) {
        val table = getTTableMap().get(schemaTableName);
        if (table == null) {
            return null;
        }
        return table.getTColumnHandles();
    }

    public TechscoreRecordCursor createTRecordCursor(@NonNull TechscoreSplit tSplit, @NonNull List<TechscoreColumnHandle> tColumnHandles) {
        val schemaTableName = tSplit.getSchemaTableName();
        val table = getTTableMap().get(schemaTableName);
        if (table == null) {
            return null;
        }
        return new TechscoreRecordCursor(tColumnHandles, table.getData());
    }
}

package com.techscore.example.presto.plugin;

import static java.util.stream.Collectors.toList;

import java.util.List;

import java.util.Map;

import javax.inject.Inject;

import lombok.Getter;

import lombok.NonNull;

import lombok.val;

import org.weakref.jmx.internal.guava.collect.Maps;

import com.facebook.presto.spi.SchemaTableName;

import com.google.inject.name.Named;

public class TechscoreClient {

@Getter

private final String connectorId;

@Getter

private final TechscoreConnectorConfig tConnectorConfig;

@Getter

private final Map<SchemaTableName, TechscoreTable> tTableMap;

@Inject

public TechscoreClient(@Named("connectorId") @NonNull String connectorId,

@NonNull TechscoreConnectorConfig tConnectorConfig) {

this.connectorId = connectorId;

this.tConnectorConfig = tConnectorConfig;

this.tTableMap = Maps.uniqueIndex(TechscoreTables.list(connectorId, tConnectorConfig.getUrlBase()), t -> t.getSchemaTableName());

}

public List<String> listSchemaNames() {

return getTTableMap().values().stream().map(TechscoreTable::getSchemaName).collect(toList());

}

public List<SchemaTableName> listSchemaTableNames() {

return getTTableMap().values().stream().map(TechscoreTable::getSchemaTableName).collect(toList());

}

public TechscoreTableHandle getTTableHandle(@NonNull SchemaTableName schemaTableName) {

val table = getTTableMap().get(schemaTableName);

if (table == null) {

return null;

}

return table.getTTableHandle();

}

public List<TechscoreColumnHandle> getTColumnHandles(@NonNull SchemaTableName schemaTableName) {

val table = getTTableMap().get(schemaTableName);

if (table == null) {

return null;

}

return table.getTColumnHandles();

}

public TechscoreRecordCursor createTRecordCursor(@NonNull TechscoreSplit tSplit, @NonNull List<TechscoreColumnHandle> tColumnHandles) {

val schemaTableName = tSplit.getSchemaTableName();

val table = getTTableMap().get(schemaTableName);

if (table == null) {

return null;

}

return new TechscoreRecordCursor(tColumnHandles, table.getData());

}

com.techscore.example.presto.plugin.TechscoreTables

package com.techscore.example.presto.plugin;

import java.time.LocalDateTime;
import java.util.List;

import lombok.NonNull;
import lombok.val;

import com.facebook.presto.spi.SchemaTableName;
import com.facebook.presto.spi.type.BigintType;
import com.facebook.presto.spi.type.TimestampType;
import com.facebook.presto.spi.type.VarcharType;
import com.google.common.collect.ImmutableList;

public class TechscoreTables {

    public static List<TechscoreTable> list(@NonNull String connectorId, String urlBase) {
        // 記事テーブル定義
        val entriesTableHandle = new TechscoreTableHandle(connectorId, new SchemaTableName("schema1", "entries"));
        val entriesColumnHandle = ImmutableList.<TechscoreColumnHandle>of(
                new TechscoreColumnHandle(connectorId, "id", BigintType.BIGINT, 0),
                new TechscoreColumnHandle(connectorId, "title", VarcharType.VARCHAR, 1),
                new TechscoreColumnHandle(connectorId, "path", VarcharType.VARCHAR, 2),
                new TechscoreColumnHandle(connectorId, "posted_by", BigintType.BIGINT, 3),
                new TechscoreColumnHandle(connectorId, "posted_at", TimestampType.TIMESTAMP, 4));
        val entriesData = ImmutableList.<List<Object>>of(
                ImmutableList.<Object>of(Long.valueOf(1L), "Riding Rails 勉強会 #1 Rails をのりこなせ！やりました！発表資料も！", urlBase + "2015/04/23/riding-rails-vol1-yattayo/", Long.valueOf(1L), LocalDateTime.parse("2015-04-23T10:54:00")),
                ImmutableList.<Object>of(Long.valueOf(2L), "Ruby on Rails 4.2 rc2 が出たので Release ノートを読む", urlBase + "2014/12/13/rails-4-2-release-notes/", Long.valueOf(1L), LocalDateTime.parse("2014-12-13T10:00:00")),
                ImmutableList.<Object>of(Long.valueOf(3L), "Gradle デフォルトタスクに名前をつける", urlBase + "2014/12/23/gradle-defaulttasks/", Long.valueOf(2L), LocalDateTime.parse("2014-12-23T10:00:00")));
        val entriesTable = new TechscoreTable(entriesTableHandle, entriesColumnHandle, entriesData);

        // 著者テーブル定義
        val authorsTableHandle = new TechscoreTableHandle(connectorId, new SchemaTableName("schema1", "authors"));
        val authorsColumnHandle = ImmutableList.<TechscoreColumnHandle>of(
                new TechscoreColumnHandle(connectorId, "id", BigintType.BIGINT, 0),
                new TechscoreColumnHandle(connectorId, "name", VarcharType.VARCHAR, 1));
        val authorsData = ImmutableList.<List<Object>>of(
                ImmutableList.<Object>of(Long.valueOf(1L), "suzuki-kei"),
                ImmutableList.<Object>of(Long.valueOf(2L), "ter@"));
        val authorsTable = new TechscoreTable(authorsTableHandle, authorsColumnHandle, authorsData);

        return ImmutableList.of(entriesTable, authorsTable);
    }
}

package com.techscore.example.presto.plugin;

import java.time.LocalDateTime;

import java.util.List;

import lombok.NonNull;

import lombok.val;

import com.facebook.presto.spi.SchemaTableName;

import com.facebook.presto.spi.type.BigintType;

import com.facebook.presto.spi.type.TimestampType;

import com.facebook.presto.spi.type.VarcharType;

import com.google.common.collect.ImmutableList;

public class TechscoreTables {

public static List<TechscoreTable> list(@NonNull String connectorId, String urlBase) {

// 記事テーブル定義

val entriesTableHandle = new TechscoreTableHandle(connectorId, new SchemaTableName("schema1", "entries"));

val entriesColumnHandle = ImmutableList.<TechscoreColumnHandle>of(

new TechscoreColumnHandle(connectorId, "id", BigintType.BIGINT, 0),

new TechscoreColumnHandle(connectorId, "title", VarcharType.VARCHAR, 1),

new TechscoreColumnHandle(connectorId, "path", VarcharType.VARCHAR, 2),

new TechscoreColumnHandle(connectorId, "posted_by", BigintType.BIGINT, 3),

new TechscoreColumnHandle(connectorId, "posted_at", TimestampType.TIMESTAMP, 4));

val entriesData = ImmutableList.<List<Object>>of(

ImmutableList.<Object>of(Long.valueOf(1L), "Riding Rails 勉強会 #1 Rails をのりこなせ！やりました！発表資料も！", urlBase + "2015/04/23/riding-rails-vol1-yattayo/", Long.valueOf(1L), LocalDateTime.parse("2015-04-23T10:54:00")),

ImmutableList.<Object>of(Long.valueOf(2L), "Ruby on Rails 4.2 rc2 が出たので Release ノートを読む", urlBase + "2014/12/13/rails-4-2-release-notes/", Long.valueOf(1L), LocalDateTime.parse("2014-12-13T10:00:00")),

ImmutableList.<Object>of(Long.valueOf(3L), "Gradle デフォルトタスクに名前をつける", urlBase + "2014/12/23/gradle-defaulttasks/", Long.valueOf(2L), LocalDateTime.parse("2014-12-23T10:00:00")));

val entriesTable = new TechscoreTable(entriesTableHandle, entriesColumnHandle, entriesData);

// 著者テーブル定義

val authorsTableHandle = new TechscoreTableHandle(connectorId, new SchemaTableName("schema1", "authors"));

val authorsColumnHandle = ImmutableList.<TechscoreColumnHandle>of(

new TechscoreColumnHandle(connectorId, "id", BigintType.BIGINT, 0),

new TechscoreColumnHandle(connectorId, "name", VarcharType.VARCHAR, 1));

val authorsData = ImmutableList.<List<Object>>of(

ImmutableList.<Object>of(Long.valueOf(1L), "suzuki-kei"),

ImmutableList.<Object>of(Long.valueOf(2L), "ter@"));

val authorsTable = new TechscoreTable(authorsTableHandle, authorsColumnHandle, authorsData);

return ImmutableList.of(entriesTable, authorsTable);

}

com.techscore.example.presto.plugin.TechscoreTable

package com.techscore.example.presto.plugin;

import java.util.List;

import lombok.NonNull;
import lombok.Value;

import org.weakref.jmx.internal.guava.collect.ImmutableList;

import com.facebook.presto.spi.SchemaTableName;

@Value
public class TechscoreTable {

    private final TechscoreTableHandle tTableHandle;
    private final List<TechscoreColumnHandle> tColumnHandles;
    private final List<List<Object>> data;

    public TechscoreTable(@NonNull TechscoreTableHandle tTableHandle,
            @NonNull List<TechscoreColumnHandle> tColumnHandle,
            @NonNull List<List<Object>> data) {
        this.tTableHandle = tTableHandle;
        this.tColumnHandles = ImmutableList.copyOf(tColumnHandle);
        this.data = ImmutableList.copyOf(data);
    }

    public SchemaTableName getSchemaTableName() {
        return getTTableHandle().getSchemaTableName();
    }

    public String getSchemaName() {
        return getTTableHandle().getSchemaName();
    }

    public String getTableName() {
        return getTTableHandle().getTableName();
    }
}

package com.techscore.example.presto.plugin;

import java.util.List;

import lombok.NonNull;

import lombok.Value;

import org.weakref.jmx.internal.guava.collect.ImmutableList;

import com.facebook.presto.spi.SchemaTableName;

@Value

public class TechscoreTable {

private final TechscoreTableHandle tTableHandle;

private final List<TechscoreColumnHandle> tColumnHandles;

private final List<List<Object>> data;

public TechscoreTable(@NonNull TechscoreTableHandle tTableHandle,

@NonNull List<TechscoreColumnHandle> tColumnHandle,

@NonNull List<List<Object>> data) {

this.tTableHandle = tTableHandle;

this.tColumnHandles = ImmutableList.copyOf(tColumnHandle);

this.data = ImmutableList.copyOf(data);

}

public SchemaTableName getSchemaTableName() {

return getTTableHandle().getSchemaTableName();

}

public String getSchemaName() {

return getTTableHandle().getSchemaName();

}

public String getTableName() {

return getTTableHandle().getTableName();

}

と、ここで気づきましたが、このままでは TechscoreTables 内の文字列リテラルが文字化けを起こしてしまいますね。build.gradle にエンコーディング設定を追加しておきましょう。

tasks.withType(AbstractCompile) each { it.options.encoding = 'UTF-8' }

1	tasks.withType(AbstractCompile) each { it.options.encoding = 'UTF-8' }

次回は

第三回の次回はいよいよレコード操作系のクラスの実装に進みます。