Arrow Flight SQLプロトコルによる接続

Doris 2.1以降、Arrow Flight SQLプロトコルに基づく高速データリンクが実装され、SQLクエリによってDorisから大量のデータを複数の言語で迅速に取得できるようになりました。Arrow Flight SQLは汎用的なJDBCドライバーも提供し、同じくArrow Flight SQLプロトコルに従うデータベースとのシームレスな連携をサポートしています。一部のシナリオでは、MySQL ClientやJDBC/ODBCドライバーを使用したデータ転送ソリューションと比較して、パフォーマンスが最大100倍向上する場合があります。

実装原理

Dorisでは、クエリ結果はBlocksとして列形式で整理されます。2.1以前のバージョンでは、データはMySQL ClientやJDBC/ODBCドライバーを介してターゲットクライアントに転送できましたが、これには行ベースのBytesを列形式に逆シリアル化する必要がありました。Arrow Flight SQLに基づく高速データ転送リンクを構築することで、ターゲットクライアントもArrow列形式をサポートしている場合、転送プロセス全体でシリアル化および逆シリアル化操作を回避し、それらに関連する時間とパフォーマンスのオーバーヘッドを完全に排除できます。

Apache Arrowをインストールするには、公式ドキュメントApache Arrowで詳細なインストール手順を確認できます。DorisがArrow Flightプロトコルを実装する方法の詳細については、Doris support Arrow Flight SQL protocolを参照してください。

Python使用方法

PythonのADBC Driverを使用してDorisに接続し、極めて高速なデータ読み取りを実現します。以下の手順では、Python（バージョン >= 3.9）ADBC Driverを使用して、DDL、DML、Session変数の設定、Show文を含む一連の一般的なデータベース構文操作を実行します。

ライブラリのインストール

このライブラリはPyPIで公開されており、以下の方法で簡単にインストールできます：

pip install adbc_driver_manager
pip install adbc_driver_flightsql

インストールされたライブラリを使用するために、コード内で以下のモジュール/ライブラリをインポートしてください：

import adbc_driver_manager
import adbc_driver_flightsql.dbapi as flight_sql

>>> print(adbc_driver_manager.__version__)
1.1.0
>>> print(adbc_driver_flightsql.__version__)
1.1.0

Dorisへの接続

Doris Arrow Flight SQLサービスとやり取りするためのクライアントを作成します。Doris FEのHost、Arrow Flight Port、ログインユーザー名とパスワードを提供し、以下の設定を実行する必要があります。 Doris FEとBEの設定パラメータを変更します：

• fe/conf/fe.confのarrow_flight_sql_portを8070などの利用可能なポートに変更します。
• be/conf/be.confのarrow_flight_sql_portを8050などの利用可能なポートに変更します。

注意: fe.confとbe.confで設定するarrow_flight_sql_portのポート番号は異なります

設定を変更してクラスターを再起動した後、fe/log/fe.logファイルでArrow Flight SQL service is startedを検索すると、FEのArrow Flight Serverが正常に開始されたことを示します。be/log/be.INFOファイルでArrow Flight Service bind to hostを検索すると、BEのArrow Flight Serverが正常に開始されたことを示します。

DorisインスタンスのFEとBEのArrow Flight SQLサービスがそれぞれポート8070と8050で実行され、Dorisのユーザー名/パスワードが"user"/"pass"であると仮定すると、接続プロセスは以下の通りです：

conn = flight_sql.connect(uri="grpc://{FE_HOST}:{fe.conf:arrow_flight_sql_port}", db_kwargs={
            adbc_driver_manager.DatabaseOptions.USERNAME.value: "user",
            adbc_driver_manager.DatabaseOptions.PASSWORD.value: "pass",
        })
cursor = conn.cursor()

接続が完了した後、返されたCursorを使用してSQLを通じてDorisと対話し、テーブルの作成、メタデータの取得、データのインポート、クエリなどの操作を実行できます。

テーブルの作成とメタデータの取得

cursor.execute()関数にQueryを渡して、テーブル作成とメタデータ取得操作を実行します：

cursor.execute("DROP DATABASE IF EXISTS arrow_flight_sql FORCE;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("create database arrow_flight_sql;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("show databases;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("use arrow_flight_sql;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("""CREATE TABLE arrow_flight_sql_test
    (
         k0 INT,
         k1 DOUBLE,
         K2 varchar(32) NULL DEFAULT "" COMMENT "",
         k3 DECIMAL(27,9) DEFAULT "0",
         k4 BIGINT NULL DEFAULT '10',
         k5 DATE,
    )
    DISTRIBUTED BY HASH(k5) BUCKETS 5
    PROPERTIES("replication_num" = "1");""")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("show create table arrow_flight_sql_test;")
print(cursor.fetchallarrow().to_pandas())

StatusResultが0を返す場合、Queryが正常に実行されたことを意味します（この設計の理由はJDBCとの互換性を保つためです）。

  StatusResult
0            0

  StatusResult
0            0

                   Database
0         __internal_schema
1          arrow_flight_sql
..                      ...
507             udf_auth_db

[508 rows x 1 columns]

  StatusResult
0            0

  StatusResult
0            0
                   Table                                       Create Table
0  arrow_flight_sql_test  CREATE TABLE `arrow_flight_sql_test` (\n  `k0`...

データのインポート

作成したテーブルに少量のテストデータをインポートするため、INSERT INtoを実行します：

cursor.execute("""INSERT INTO arrow_flight_sql_test VALUES
        ('0', 0.1, "ID", 0.0001, 9999999999, '2023-10-21'),
        ('1', 0.20, "ID_1", 1.00000001, 0, '2023-10-21'),
        ('2', 3.4, "ID_1", 3.1, 123456, '2023-10-22'),
        ('3', 4, "ID", 4, 4, '2023-10-22'),
        ('4', 122345.54321, "ID", 122345.54321, 5, '2023-10-22');""")
print(cursor.fetchallarrow().to_pandas())

以下により、インポートが成功したことが証明されます：

  StatusResult
0            0

Dorisに大量のデータをインポートする必要がある場合は、pydorisを使用してStream Loadを実行できます。

クエリの実行

次に、上記でインポートしたテーブルに対してクエリを実行します。これには集約、ソート、Set Session Variableなどの操作が含まれます。

cursor.execute("select * from arrow_flight_sql_test order by k0;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("set exec_mem_limit=2000;")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("show variables like \"%exec_mem_limit%\";")
print(cursor.fetchallarrow().to_pandas())

cursor.execute("select k5, sum(k1), count(1), avg(k3) from arrow_flight_sql_test group by k5;")
print(cursor.fetch_df())

結果は以下の通りです：

   k0            k1    K2                k3          k4          k5
0   0       0.10000    ID       0.000100000  9999999999  2023-10-21
1   1       0.20000  ID_1       1.000000010           0  2023-10-21
2   2       3.40000  ID_1       3.100000000      123456  2023-10-22
3   3       4.00000    ID       4.000000000           4  2023-10-22
4   4  122345.54321    ID  122345.543210000           5  2023-10-22

[5 rows x 6 columns]

  StatusResult
0            0

    Variable_name Value Default_Value Changed
0  exec_mem_limit  2000    2147483648       1

           k5  Nullable(Float64)_1  Int64_2 Nullable(Decimal(38, 9))_3
0  2023-10-22         122352.94321        3            40784.214403333
1  2023-10-21              0.30000        2                0.500050005

[2 rows x 5 columns]

注意: クエリ結果を取得するには、arrow形式を返すcursor.fetchallarrow()を使用するか、pandas dataframeを直接返すcursor.fetch_df()を使用する必要があります。これによりデータがカラム形式で保持されます。cursor.fetchall()は使用しないでください。そうしないと、カラム形式のデータが行形式に変換されてしまい、本質的にはmysql-clientを使用するのと同じになります。実際、クライアント側での余分なカラムから行への変換操作により、mysql-clientよりも遅くなる可能性があります。

完全なコード

# Doris Arrow Flight SQL Test

# step 1, library is released on PyPI and can be easily installed.
# pip install adbc_driver_manager
# pip install adbc_driver_flightsql
import adbc_driver_manager
import adbc_driver_flightsql.dbapi as flight_sql

# step 2, create a client that interacts with the Doris Arrow Flight SQL service.
# Modify arrow_flight_sql_port in fe/conf/fe.conf to an available port, such as 8070.
# Modify arrow_flight_sql_port in be/conf/be.conf to an available port, such as 8050.
conn = flight_sql.connect(uri="grpc://{FE_HOST}:{fe.conf:arrow_flight_sql_port}", db_kwargs={
            adbc_driver_manager.DatabaseOptions.USERNAME.value: "root",
            adbc_driver_manager.DatabaseOptions.PASSWORD.value: "",
        })
cursor = conn.cursor()

# interacting with Doris via SQL using Cursor
def execute(sql):
    print("\n### execute query: ###\n " + sql)
    cursor.execute(sql)
    print("### result: ###")
    print(cursor.fetchallarrow().to_pandas())

# step3, execute DDL statements, create database/table, show stmt.
execute("DROP DATABASE IF EXISTS arrow_flight_sql FORCE;")
execute("show databases;")
execute("create database arrow_flight_sql;")
execute("show databases;")
execute("use arrow_flight_sql;")
execute("""CREATE TABLE arrow_flight_sql_test
    (
         k0 INT,
         k1 DOUBLE,
         K2 varchar(32) NULL DEFAULT "" COMMENT "",
         k3 DECIMAL(27,9) DEFAULT "0",
         k4 BIGINT NULL DEFAULT '10',
         k5 DATE,
    )
    DISTRIBUTED BY HASH(k5) BUCKETS 5
    PROPERTIES("replication_num" = "1");""")
execute("show create table arrow_flight_sql_test;")


# step4, insert into
execute("""INSERT INTO arrow_flight_sql_test VALUES
        ('0', 0.1, "ID", 0.0001, 9999999999, '2023-10-21'),
        ('1', 0.20, "ID_1", 1.00000001, 0, '2023-10-21'),
        ('2', 3.4, "ID_1", 3.1, 123456, '2023-10-22'),
        ('3', 4, "ID", 4, 4, '2023-10-22'),
        ('4', 122345.54321, "ID", 122345.54321, 5, '2023-10-22');""")


# step5, execute queries, aggregation, sort, set session variable
execute("select * from arrow_flight_sql_test order by k0;")
execute("set exec_mem_limit=2000;")
execute("show variables like \"%exec_mem_limit%\";")
execute("select k5, sum(k1), count(1), avg(k3) from arrow_flight_sql_test group by k5;")

# step6, close cursor 
cursor.close()

Arrow Flight SQLを使用したJdbcコネクタ

Arrow Flight SQLプロトコルのオープンソースJDBCドライバーは標準JDBC APIと互換性があり、ほとんどのBIツールがJDBCを通じてDorisにアクセスするために使用でき、Apache Arrowデータの高速伝送をサポートします。使用方法はMySQLプロトコルのJDBCドライバーを通じてDorisに接続する場合と似ています。リンクURL内のjdbc:mysqlプロトコルをjdbc:arrow-flight-sqlプロトコルに置き換えるだけです。クエリ結果は引き続きJDBC ResultSetデータ構造で返されます。

POM依存関係：

<properties>
    <arrow.version>17.0.0</arrow.version>
</properties>
<dependencies>
    <dependency>
        <groupId>org.apache.arrow</groupId>
        <artifactId>flight-sql-jdbc-core</artifactId>
        <version>${arrow.version}</version>
    </dependency>
</dependencies>

注意: Java 9以降を使用する場合、Javaコマンドに--add-opens=java.base/java.nio=ALL-UNNAMEDを追加してJDKの内部構造を公開する必要があります。そうしないと、module java.base does not "opens java.nio" to unnamed moduleやmodule java.base does not "opens java.nio" to org.apache.arrow.memory.coreやjava.lang.NoClassDefFoundError: Could not initialize class org.apache.arrow.memory.util.MemoryUtil (Internal; Prepare)などのエラーが発生する可能性があります。

# Directly on the command line
$ java --add-opens=java.base/java.nio=ALL-UNNAMED -jar ...
# Indirectly via environment variables
$ env _JAVA_OPTIONS="--add-opens=java.base/java.nio=ALL-UNNAMED" java -jar ...

IntelliJ IDEAでデバッグを行う場合、Run/Debug ConfigurationsのBuild and runに--add-opens=java.base/java.nio=ALL-UNNAMEDを追加する必要があります。下図を参照してください：

接続コードの例は以下の通りです：

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

Class.forName("org.apache.arrow.driver.jdbc.ArrowFlightJdbcDriver");
String DB_URL = "jdbc:arrow-flight-sql://{FE_HOST}:{fe.conf:arrow_flight_sql_port}?useServerPrepStmts=false"
        + "&cachePrepStmts=true&useSSL=false&useEncryption=false";
String USER = "root";
String PASS = "";

Connection conn = DriverManager.getConnection(DB_URL, USER, PASS);
Statement stmt = conn.createStatement();
ResultSet resultSet = stmt.executeQuery("select * from information_schema.tables;");
while (resultSet.next()) {
    System.out.println(resultSet.toString());
}

resultSet.close();
stmt.close();
conn.close();

Java使用方法

JDBCを使用することに加えて、Pythonと同様に、JAVAでもDriverを作成してDorisを読み取り、Arrow形式でデータを返すことができます。以下は、AdbcDriverとJdbcDriverを使用してDoris Arrow Flight Serverに接続する方法です。

POM依存関係：

<properties>
    <adbc.version>0.15.0</adbc.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.arrow.adbc</groupId>
        <artifactId>adbc-driver-jdbc</artifactId>
        <version>${adbc.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.arrow.adbc</groupId>
        <artifactId>adbc-core</artifactId>
        <version>${adbc.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.arrow.adbc</groupId>
        <artifactId>adbc-driver-manager</artifactId>
        <version>${adbc.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.arrow.adbc</groupId>
        <artifactId>adbc-sql</artifactId>
        <version>${adbc.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.arrow.adbc</groupId>
        <artifactId>adbc-driver-flight-sql</artifactId>
        <version>${adbc.version}</version>
    </dependency>
</dependencies>

ADBC Driver

接続コードの例は以下の通りです：

// 1. new driver
final BufferAllocator allocator = new RootAllocator();
FlightSqlDriver driver = new FlightSqlDriver(allocator);
Map<String, Object> parameters = new HashMap<>();
AdbcDriver.PARAM_URI.set(parameters, Location.forGrpcInsecure("{FE_HOST}", {fe.conf:arrow_flight_sql_port}).getUri().toString());
AdbcDriver.PARAM_USERNAME.set(parameters, "root");
AdbcDriver.PARAM_PASSWORD.set(parameters, "");
AdbcDatabase adbcDatabase = driver.open(parameters);

// 2. new connection
AdbcConnection connection = adbcDatabase.connect();
AdbcStatement stmt = connection.createStatement();

// 3. execute query
stmt.setSqlQuery("select * from information_schema.tables;");
QueryResult queryResult = stmt.executeQuery();
ArrowReader reader = queryResult.getReader();

// 4. load result
List<String> result = new ArrayList<>();
while (reader.loadNextBatch()) {
    VectorSchemaRoot root = reader.getVectorSchemaRoot();
    String tsvString = root.contentToTSVString();
    result.add(tsvString);
}
System.out.printf("batchs %d\n", result.size());

// 5. close
reader.close();
queryResult.close();
stmt.close();
connection.close();

JDBC Driver

Java 9以降を使用する場合、java コマンドに --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED を追加することで、一部のJDK内部を公開する必要があります：

# Directly on the command line
$ java --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED -jar ...
# Indirectly via environment variables
$ env _JAVA_OPTIONS="--add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED" java -jar ...

そうでない場合、module java.base does not "opens java.nio" to unnamed moduleやmodule java.base does not "opens java.nio" to org.apache.arrow.memory.core、またはava.lang.NoClassDefFoundError: Could not initialize class org.apache.arrow.memory.util.MemoryUtil (Internal; Prepare)などのエラーが表示される可能性があります。

IntelliJ IDEAでデバッグする場合は、Run/Debug ConfigurationsのBuild and runに--add-opens=java.base/java.nio=ALL-UNNAMEDを追加する必要があります。以下の画像を参照してください：

接続コードの例は以下のとおりです：

final Map<String, Object> parameters = new HashMap<>();
AdbcDriver.PARAM_URI.set(
        parameters,"jdbc:arrow-flight-sql://{FE_HOST}:{fe.conf:arrow_flight_sql_port}?useServerPrepStmts=false&cachePrepStmts=true&useSSL=false&useEncryption=false");
AdbcDriver.PARAM_USERNAME.set(parameters, "root");
AdbcDriver.PARAM_PASSWORD.set(parameters, "");
try (
        BufferAllocator allocator = new RootAllocator();
        AdbcDatabase db = new JdbcDriver(allocator).open(parameters);
        AdbcConnection connection = db.connect();
        AdbcStatement stmt = connection.createStatement()
) {
    stmt.setSqlQuery("select * from information_schema.tables;");
    AdbcStatement.QueryResult queryResult = stmt.executeQuery();
    ArrowReader reader = queryResult.getReader();
    List<String> result = new ArrayList<>();
    while (reader.loadNextBatch()) {
        VectorSchemaRoot root = reader.getVectorSchemaRoot();
        String tsvString = root.contentToTSVString();
        result.add(tsvString);
    }
    long etime = System.currentTimeMillis();
    System.out.printf("batchs %d\n", result.size());

    reader.close();
    queryResult.close();
    stmt.close();
}  catch (Exception e) {
    e.printStackTrace();
}

Jdbc と Java 接続方法の選択

JDBC/Java Arrow Flight SQL Sample は、Arrow FLight SQL を使用した JDBC/Java デモです。これを使用して Arrow Flight Server にクエリを送信するための様々な接続方法をテストでき、Arrow FLight SQL の使用方法の理解とパフォーマンステストに役立ちます。期待される実行結果については、Add Arrow Flight Sql demo for Java を参照してください。

従来の jdbc:mysql 接続方法と比較した Jdbc と Java の Arrow Flight SQL 接続方法のパフォーマンステストは、GitHub Issue のセクション 6.2 で確認できます。テスト結果に基づく使用方法の提案を以下に示します。

1. 上記の3つの Java Arrow Flight SQL 接続方法について、後続のデータ分析が行ベースのデータ形式に基づいて行われる場合は、JDBC ResultSet 形式でデータを返す jdbc:arrow-flight-sql を使用してください。後続のデータ分析が Arrow 形式またはその他の列ベースのデータ形式に基づいて実行できる場合は、Flight AdbcDriver または Flight JdbcDriver を使用して Arrow 形式で直接データを返すことで、行列変換を回避し、Arrow の特性を利用してデータ解析を高速化できます。

2. JDBC ResultSet または Arrow 形式でのデータ解析において、データ解析にかかる時間はデータ読み取りにかかる時間よりも長くなります。Arrow Flight SQL のパフォーマンスが期待通りでなく、jdbc:mysql:// と比較して改善が限定的な場合は、データ解析に時間がかかりすぎているかどうかを分析することをお勧めします。

3. すべての接続方法において、JDK 17 は JDK 1.8 よりもデータ読み取りが高速です。

4. 大量のデータを読み取る際、Arrow Flight SQL は jdbc:mysql:// よりもメモリ使用量が少なくなるため、メモリ不足に悩んでいる場合は Arrow Flight SQL を試すこともできます。

5. 上記の3つの接続方法に加えて、ネイティブ FlightClient を使用して Arrow Flight Server に接続することも可能で、複数の endpoint をより柔軟に並列読み取りできます。Flight AdbcDriver も FlightClient に基づいて作成されたリンクで、FlightClient を直接使用するよりも簡単です。

他のビッグデータコンポーネントとの連携

Spark & Flink

Arrow Flight は現在、Spark と Flink をサポートする公式計画がありません（GitHub Issue）。バージョン 24.0.0 以降、Doris 独自の Spark Connector と Flink Connector が Arrow Flight SQL 経由での Doris へのアクセスをサポートしており、これにより読み取りパフォーマンスが数倍向上することが期待されます。

コミュニティでは以前、オープンソースの Spark-Flight-Connector を参照し、Spark で FlightClient を使用して Doris に接続してテストを行いました。Arrow と Doris Block 間のデータ形式変換がより高速であることが判明し、これは CSV 形式と Doris Block 間の変換速度の10倍であり、Map や Array などの複合型により良いサポートを提供します。これは Arrow データ形式が高い圧縮率と伝送時の低いネットワークオーバーヘッドを持つためです。ただし、Doris Arrow Flight はまだマルチノード並列読み取りを実装しておらず、依然としてクエリ結果を BE ノードに集約して返します。単純なデータのバッチエクスポートの場合、Tablet レベルの並列読み取りをサポートする Doris Spark Connector ほど高速ではない可能性があります。Spark で Arrow Flight SQL を使用して Doris に接続したい場合は、オープンソースの Spark-Flight-Connector と Dremio-Flight-Connector を参照して独自に実装できます。

BI ツールのサポート

Doris v2.1.8 以降、DBeaver などの BI ツールが arrow-flight-sql プロトコルを使用して Doris に接続することがサポートされています。DBeaver で arrow-flight-sql Driver を使用して Doris に接続する方法については、how-to-use-jdbc-driver-with-dbeaver-client、client-applications/clients/dbeaver/ を参照してください。

拡張アプリケーション

複数の BE による並列結果返却

Doris はデフォルトで、すべての BE ノードでのクエリ結果を1つの BE ノードに集約します。Mysql/JDBC クエリでは、FE はこの集約されたデータノードからクエリ結果を要求します。Arrow Flight SQL クエリでは、FE はこのノードの IP/Port を Endpoint にラップして ADBC Client に返します。ADBC Client はこの Endpoint に対応する BE ノードに要求してデータをプルします。

クエリが Join、Sort、Window Function などのデータ Shuffle 動作を持つ演算子を使用せず、Doris からデータをプルする単純な Select である場合、クエリは Tablet 粒度で分割できます。現在 Doris Spark/Flink Connector はこの方法を使用して並列データ読み取りを実装しており、2つのステップに分かれています：

1. explain sql を実行し、FE から返されるクエリプランの ScanOperator にスキャンのすべての Tablet ID List が含まれます。
2. 上記の Tablet ID List に基づいて元の SQL を複数の SQL に分割します。各 SQL は Tablet の一部のみを読み取ります。使用方法は SELECT * FROM t1 TABLET(10001,10002) limit 1000; と同様です。分割後の複数の SQL は並列実行できます。Support select table sample を参照してください。

クエリの最外層が集約である場合、SQL は select k1, sum(k2) from xxx group by k1 のようになります。Doris v3.0.4 以降、set enable_parallel_result_sink=true; を実行してクエリの各 BE ノードがクエリ結果を独立して返すことを許可します。FE から返された Endpoint リストを受信後、ADBC Client は複数の BE ノードから並列で結果をプルします。ただし、集約結果が非常に小さい場合、複数の BE を返すことで RPC の負荷が増加することに注意してください。具体的な実装については、support parallel result sink を参照してください。理論的には、最外層のクエリがソートされている場合を除き、他のクエリは各 BE ノードが並列で結果を返すことをサポートできますが、現在この利便性の必要がないため、さらなる実装は行われていません。

複数の BE がクラスター外からアクセス可能な同一 IP を共有

Doris クラスターがあり、その FE ノードはクラスター外からアクセス可能で、すべての BE ノードはクラスター内からのみアクセス可能な場合を考えます。Mysql Client と JDBC を使用して Doris に接続してクエリを実行する場合は問題なく、クエリ結果は Doris FE ノードから返されます。しかし、Arrow Flight SQL を使用して Doris に接続する場合、ADBC Client が Doris BE ノードに接続してクエリ結果をプルする必要がありますが、Doris BE ノードはクラスター外からのアクセスが許可されていないため機能しません。

本番環境では、Doris BE ノードをクラスター外に公開することが不便な場合がよくあります。しかし、すべての Doris BE ノードにリバースプロキシ（Nginx など）を追加できます。クラスター外のクライアントが Nginx に接続すると、Doris BE ノードにランダムにルーティングされます。デフォルトでは、Arrow Flight SQL クエリ結果はランダムに Doris BE ノードに保存されます。Nginx によってランダムにルーティングされた Doris BE ノードと異なる場合、Doris BE ノード内でデータ転送が必要になります。

Doris v2.1.8 以降、すべての Doris BE ノードの be.conf で public_host と arrow_flight_sql_proxy_port を、複数の Doris BE ノードが共有しクラスター外からアクセス可能な IP とポートに設定することで、クエリ結果を正しく転送して ADBC Client に返すことができます。

public_host={nginx ip}
arrow_flight_sql_proxy_port={nginx port}

FAQ

1. Q: エラー connection error: desc = "transport: Error while dialing: dial tcp <ip:arrow_flight_port>: i/o timeout"。

A: エラーメッセージの <ip:arrow_flight_port> が Doris FE ノードの IP と arrow-flight-port の場合、

まず Doris FE ノードの arrow-flight-server が正常に起動しているかを確認してください。fe/log/fe.log ファイルで Arrow Flight SQL service is started を検索することで、FE の Arrow Flight Server が正常に起動していることが確認できます。

Doris FE ノードの arrow-flight-server が正常に起動している場合、Client が配置されているマシンからエラーメッセージ内の IP <ip:arrow_flight_port> に ping が可能かを確認してください。ping できない場合は、Doris FE ノードに外部からアクセス可能な IP を開放し、クラスターを再デプロイする必要があります。

A: エラーメッセージの <ip:arrow_flight_port> が Doris BE ノードの IP と arrow-flight-port の場合。

まず Doris BE ノードの arrow-flight-server が正常に起動しているかを確認してください。be/log/be.INFO ファイルで Arrow Flight Service bind to host を検索することで、BE の Arrow Flight Server が正常に起動していることが確認できます。

Doris BE ノードの arrow-flight-server が正常に起動している場合、クライアントマシンからエラーメッセージで報告された <ip:arrow_flight_port> の IP に ping が可能かを確認してください。ping できない場合で、Doris BE ノードが外部からアクセスできない内部ネットワークにあることが既知の場合、以下の2つの方法を使用してください：

• 各 Doris BE ノードに外部からアクセス可能な IP を開放することを検討してください。Doris v2.1.8 以降では、この Doris BE ノードの be.conf でこの IP に public_host を設定できます。同様に、すべての Doris BE ノードの public_host を、クライアントからアクセス可能な対応する BE ノードの IP に設定してください。

• 上記のセクション [クラスターで外部からアクセス可能な同一 IP を複数の BE が共有] を参照して、すべての Doris BE ノードにリバースプロキシの層を追加してください。

Doris BE が完全に内部ネットワークにあるかどうかが不明な場合は、クライアントマシンと Doris BE ノードが配置されているマシンの他の IP との接続性を確認してください。Doris BE ノードが配置されているマシンで ifconfig を実行すると、現在のマシンのすべての IP が返されます。IP の1つは <ip:arrow_flight_port> の IP と同じで、show backends で出力される Doris BE ノードの IP と同じである必要があります。ifconfig で返される他の IP を順次 ping してください。Doris BE ノードに Client からアクセス可能な IP がある場合は、上記を参照してこの IP を public_host として設定してください。Doris BE ノードのすべての IP が Client からアクセスできない場合、Doris BE ノードは完全に内部ネットワークにあります。

2. Q: JDBC または JAVA を使用して Arrow Flight SQL に接続する際に、エラーメッセージが表示される：module java.base does not "opens java.nio" to unnamed module または module java.base does not "opens java.nio" to org.apache.arrow.memory.core または java.lang.NoClassDefFoundError: Could not initialize class org.apache.arrow.memory.util.MemoryUtil (Internal; Prepare)

A: まず fe/conf/fe.conf の JAVA_OPTS_FOR_JDK_17 に --add-opens=java.base/java.nio=ALL-UNNAMED が含まれているかを確認してください。含まれていない場合は追加してください。次に、上記の [JDBC Connector with Arrow Flight SQL] の注意事項を参照し、Java コマンドに --add-opens=java.base/java.nio=ALL-UNNAMED を追加してください。IntelliJ IDEA でデバッグする場合は、Run/Debug Configurations の Build and run に --add-opens=java.base/java.nio=ALL-UNNAMED を追加する必要があります。

3. Q: ARM 環境でエラー get flight info statement failed, arrow flight schema timeout, TimeoutException: Waited 5000 milliseconds for io.grpc.stub.Client が報告される。

A: Linux カーネルバージョンが <= 4.19.90 の場合、4.19.279 以上にアップグレードするか、低いバージョンの Linux カーネル環境で Doris BE を再コンパイルする必要があります。具体的なコンパイル方法については、ドキュメント <docs/dev/install/source-install/compilation-arm> を参照してください

原因：これは古いバージョンの Linux カーネルと Arrow の間の互換性問題です。cpp: arrow::RecordBatch::MakeEmpty() が Arrow Record Batch を構築する際にスタックし、Doris BE の Arrow Flight Server が Doris FE の Arrow Flight Server の RPC リクエストに 5000ms 以内に応答できず、FE が Client に rpc timeout failed を返すことが原因です。Spark と Flink が Doris を読み取る際も、クエリ結果を Arrow Record Batch に変換して返すため、同じ問題が存在します。

kylinv10 SP2 と SP3 の Linux カーネルバージョンは最大でも 4.19.90-24.4.v2101.ky10.aarch64 のみです。カーネルバージョンをさらにアップグレードすることはできません。Doris BE は kylinv10 上でのみ再コンパイルできます。新しいバージョンの ldb_toolchain で Doris BE をコンパイルしても問題が残る場合は、低いバージョンの ldb_toolchain v0.17 でコンパイルを試してください。ARM 環境が外部ネットワークに接続できない場合、Huawei Cloud は ARM + kylinv10 を、Alibaba Cloud は x86 + kylinv10 を提供しています

4. Q: Prepared statement でパラメータを渡すとエラーが報告される。

A: 現在、jdbc:arrow-flight-sql と Java ADBC/JDBCDriver は prepared statement のパラメータ渡しをサポートしていません。例えば、select * from xxx where id=? は parameter ordinal 1 out of range エラーを報告します。これは Arrow Flight SQL のバグです（GitHub Issue）。

5. Q: 一部のシナリオでパフォーマンスを向上させるために、jdbc:arrow-flight-sql が毎回読み取るバッチサイズを変更する方法。

A: org.apache.arrow.adbc.driver.jdbc.JdbcArrowReader ファイルの makeJdbcConfig メソッド内の setTargetBatchSize を変更し（デフォルトは 1024）、変更したファイルを同じパス名でローカルディレクトリに保存して、元のファイルを上書きして有効にします。

6. Q: ADBC v0.10、JDBC、Java ADBC/JDBCDriver は並列読み取りをサポートしていない。

A: stmt.executePartitioned() メソッドが実装されていません。ネイティブ FlightClient のみを使用して複数エンドポイントの並列読み取りを実装できます。方法は sqlClient=new FlightSqlClient, execute=sqlClient.execute(sql), endpoints=execute.getEndpoints(), for(FlightEndpoint endpoint: endpoints) です。また、ADBC V0.10 のデフォルト AdbcStatement は実際には JdbcStatement です。executeQuery 後、行形式の JDBC ResultSet が Arrow カラム形式に変換されます。Java ADBC は ADBC 1.0.0 までに完全に機能することが期待されています GitHub Issue。

7. Q: URL でデータベース名を指定する。

A: Arrow v15.0 現在、Arrow JDBC Connector は URL でのデータベース名指定をサポートしていません。例えば、jdbc:arrow-flight-sql://{FE_HOST}:{fe.conf:arrow_flight_sql_port}/test?useServerPrepStmts=false で test データベースへの接続を指定しても無効で、手動で SQL use database を実行する必要があります。Arrow v18.0 は URL でのデータベース名指定をサポートしていますが、実際のテストではまだバグが存在します。

8. Q: Python ADBC で Warning: Cannot disable autocommit; conn will not be DB-API 2.0 compliant が出力される。

A: Python 使用時はこの Warning を無視してください。これは Python ADBC Client の問題で、クエリには影響しません。

9. Q: Python でエラー grpc: received message larger than max (20748753 vs. 16777216) が報告される。

A: Python: grpc: received message larger than max (20748753 vs. 16777216) #2078 を参照し、Database Option に adbc_driver_flightsql.DatabaseOptions.WITH_MAX_MSG_SIZE.value を追加してください。

10. Q: エラー invalid bearer token が報告される。

A: SET PROPERTY FOR 'root' 'max_user_connections' = '10000'; を実行して、現在のユーザーの現在の最大接続数を 10000 に変更し、fe.conf に qe_max_connection=30000 と arrow_flight_token_cache_size=8000 を追加して FE を再起動してください。

ADBC Client と Arrow Flight Server 間の接続は本質的に長時間接続で、Server 上で Auth Token、Connection、Session のキャッシュが必要です。接続が作成された後、単一クエリの終了時にすぐに切断されることはありません。Client が close() リクエストを送信してクリーンアップする必要がありますが、実際には Client が close リクエストを送信しないことが多いため、Auth Token、Connection、Session は Arrow Flight Server 上に長時間保存されます。デフォルトでは 3日後にタイムアウトして切断されるか、接続数が arrow_flight_token_cache_size の制限を超えた後に LRU に従って削除されます。

Doris v2.1.8 現在、Arrow Flight 接続と Mysql/JDBC 接続は同じ接続制限を使用し、すべての FE ユーザーの総接続数 qe_max_connection と UserProperty 内の単一ユーザーの接続数 max_user_connections を含みます。しかし、デフォルトの qe_max_connection と max_user_connections はそれぞれ 1024 と 100 です。Arrow Flight SQL は JDBC シナリオの代替としてよく使用されますが、JDBC 接続はクエリ終了後すぐに解放されます。そのため、Arrow Flight SQL を使用する際、Doris のデフォルト接続制限は小さすぎ、接続数が arrow_flight_token_cache_size の制限を超えて使用中の接続が削除されることがよくあります。

11. Q: JDBC または JAVA を使用して Arrow Flight SQL に接続し Datatime 型を読み取ると、フォーマットされた時刻ではなくタイムスタンプが返される。

A: JDBC または JAVA を使用して Arrow Flight SQL に接続し Datatime 型を読み取る場合、タイムスタンプを自分で変換する必要があります。Add java parsing datetime type in arrow flight sql sample #48578 を参照してください。Python Arrow Flight SQL を使用して Datatime 型を読み取ると 2025-03-03 17:23:28Z の結果が返されますが、JDBC または JAVA は 1740993808 を返します。

12. Q: JDBC または Java JDBC Client を使用して Arrow Flight SQL に接続し Array ネスト型を読み取るとエラー Configuration does not provide a mapping for array column 2 が返される。

A: sample/arrow-flight-sql を参照して JAVA ADBC Client を使用してください。

Python ADBC Client、JAVA ADBC Client、Java JDBC DriverManager はすべて Array ネスト型の読み取りに対応しています。JDBC または Java JDBC Client を使用して Arrow Flight SQL に接続する場合のみ問題があります。実際、Arrow Flight JDBC の互換性は保証されていません。これは Arrow が公式に開発したものではなく、サードパーティのデータベース企業 Dremio によるものです。以前にも他の互換性問題が発見されているため、まず JAVA ADBC Client の使用を推奨します。

2.1 Release Note

Doris Arrow Flight はバージョン v2.1.4 以前では完全ではないため、使用前のアップグレードを推奨します。

v2.1.9

1. Doris データの Arrow へのシリアライゼーション問題を修正。 Fix UT DataTypeSerDeArrowTest of Array/Map/Struct/Bitmap/HLL/Decimal256 types

• Decimal256 型の読み取りに失敗；
• DatetimeV2 型の読み取りでの微細なエラー；
• DateV2 型の読み取りでの不正な結果；
• IPV4/IPV6 型の結果が NULL の場合のエラー；

2. Doris Arrow Flight SQL クエリが失敗して空の結果を返し、実際のエラー情報が返されない問題を修正。 Fix query result is empty and not return query error message

v2.1.8

1. DBeaver などの BI ツールが arrow-flight-sql プロトコルを使用して Doris に接続し、メタデータツリーの正しい表示をサポート。 Support arrow-flight-sql protocol getStreamCatalogs, getStreamSchemas, getStreamTables #46217。

2. 複数の BE がクラスター外部からアクセス可能な同一 IP を共有する場合、クエリ結果を正しく転送して ADBC Client に返すことが可能。 Arrow flight server supports data forwarding when BE uses public vip

3. 複数エンドポイントの並列読み取りをサポート。 Arrow Flight support multiple endpoints

4. クエリエラー FE not found arrow flight schema を修正。 Fix FE not found arrow flight schema

5. NULL を許可するカラムを読み取る際のエラー BooleanBuilder::AppendValues を修正。 Fix Doris NULL column conversion to arrow batch

6. show processlist で重複する Connection ID が表示される問題を修正。 Fix arrow-flight-sql ConnectContext to use a unified ID #46284

7. Datetime と DatetimeV2 型を読み取る際にタイムゾーンが失われ、実際のデータより 8時間少ない datetime になる問題を修正。 Fix time zone issues and accuracy issues #38215

v2.1.7

1. 頻繁なログ出力 Connection wait_timeout を修正。 Fix kill timeout FlightSqlConnection and FlightSqlConnectProcessor close

2. Arrow Flight Bearer Token が Cache から期限切れで削除される問題を修正。 Fix Arrow Flight bearer token cache evict after expired

v2.1.6

1. クエリエラー 0.0.0.0:xxx, connection refused を修正。 Fix return result from FE Arrow Flight server error 0.0.0.0:xxx, connection refused

2. クエリエラー Reach limit of connections を修正。 Fix exceed user property max connection cause Reach limit of connections #39127

以前のバージョンでは、SET PROPERTY FOR 'root' 'max_user_connections' = '1024'; を実行して現在のユーザーの現在の最大接続数を 1024 に変更することで、一時的に回避できます。

以前のバージョンは Arrow Flight 接続数を qe_max_connection/2 未満に制限するのみで、qe_max_connection はすべての fe ユーザーの総接続数でデフォルトは 1024 ですが、単一ユーザーの Arrow Flight 接続数を UserProperty の max_user_connections 未満（デフォルトは 100）に制限しないため、Arrow Flight 接続数が現在のユーザーの接続数上限を超えるとエラー Reach limit of connections が報告されるので、現在のユーザーの max_user_connections を増加させる必要があります。

問題の詳細については、以下を参照してください：Questions

3. 1回に返されるクエリ結果の ArrowBatch サイズを変更するための Conf arrow_flight_result_sink_buffer_size_rows を追加、デフォルトは 4096 * 8。 Add config arrow_flight_result_sink_buffer_size_rows

v2.1.5

1. Arrow Flight SQL クエリ結果が空になる問題を修正。 Fix arrow flight result sink #36827

Doris v2.1.4 では大量データを読み取る際にエラーが報告される可能性があります。詳細については以下を参照してください：Questions

3.0 Release Note

v3.0.5

1. Doris データの Arrow へのシリアライゼーション問題を修正。 Fix UT DataTypeSerDeArrowTest of Array/Map/Struct/Bitmap/HLL/Decimal256 types

• Decimal256 型の読み取りに失敗；
• DatetimeV2 型の読み取りでの微細なエラー；
• DateV2 型の読み取りでの不正な結果；
• IPV4/IPV6 型の結果が NULL の場合のエラー；

v3.0.4

1. DBeaver などの BI ツールが arrow-flight-sql プロトコルを使用して Doris に接続し、メタデータツリーの正しい表示をサポート。 Support arrow-flight-sql protocol getStreamCatalogs, getStreamSchemas, getStreamTables #46217。

2. 複数エンドポイントの並列読み取りをサポート。 Arrow Flight support multiple endpoints

3. NULL を許可するカラムを読み取る際のエラー BooleanBuilder::AppendValues を修正。 Fix Doris NULL column conversion to arrow batch

4. show processlist で重複する Connection ID が表示される問題を修正。 Fix arrow-flight-sql ConnectContext to use a unified ID #46284

5. Doris Arrow Flight SQL クエリが失敗して空の結果を返し、実際のエラー情報が返されない問題を修正。 Fix query result is empty and not return query error message

v3.0.3

1. クエリエラー 0.0.0.0:xxx, connection refused を修正。 Fix return result from FE Arrow Flight server error 0.0.0.0:xxx, connection refused

2. クエリエラー Reach limit of connections を修正。 Fix exceed user property max connection cause Reach limit of connections #39127

以前のバージョンでは、SET PROPERTY FOR 'root' 'max_user_connections' = '1024'; を実行して現在のユーザーの現在の最大接続数を 1024 に変更することで、一時的に回避できます。

以前のバージョンは Arrow Flight 接続数を qe_max_connection/2 未満に制限するのみで、qe_max_connection はすべての fe ユーザーの総接続数でデフォルトは 1024 ですが、単一ユーザーの Arrow Flight 接続数を UserProperty の max_user_connections 未満（デフォルトは 100）に制限しないため、Arrow Flight 接続数が現在のユーザーの接続数上限を超えるとエラー Reach limit of connections が報告されるので、現在のユーザーの max_user_connections を増加させる必要があります。

問題の詳細については、以下を参照してください：Questions

3. 頻繁なログ出力 Connection wait_timeout を修正。 Fix kill timeout FlightSqlConnection and FlightSqlConnectProcessor close

4. Arrow Flight Bearer Token が期限切れ後に Cache から削除される問題を修正。 Fix Arrow Flight bearer token cache evict after expired

5. 複数の BE がクラスター外部からアクセス可能な同一 IP アドレスを共有する場合、クエリ結果を正しく転送して ADBC Client に返すことが可能。 Arrow flight server supports data forwarding when BE uses public vip

6. クエリエラー FE not found arrow flight schema を修正。 Fix FE not found arrow flight schema

7. Datetime と DatetimeV2 型を読み取る際にタイムゾーンが失われ、実際のデータより 8時間少ない datetime になる問題を修正。 Fix time zone issues and accuracy issues #38215

v3.0.2

1. 1回のトランザクションで返されるクエリ結果の ArrowBatch サイズを変更するための Conf arrow_flight_result_sink_buffer_size_rows を追加、デフォルトは 4096 * 8。 Add config arrow_flight_result_sink_buffer_size_rows

v3.0.1

1. クエリ結果の欠落、クエリ結果行数 = 実際の行数 / BE 数 Fix get Schema failed when enable_parallel_result_sink is false #37779

Doris 3.0.0 では、クエリの最外層が集約の場合、SQL が select k1, sum(k2) from xxx group by k1 のような場合に、（クエリ結果行数 = 実際の行数 / BE 数）の問題に遭遇する可能性があります。これは support parallel result sink によって導入された問題です。Fix get Schema failed when enable_parallel_result_sink is false は一時的な修正で、Arrow Flight support multiple endpoints が複数エンドポイントの並列読み取りをサポートした後に正式に修正される予定です。