BigQuery Java 客户端：如何高效管理会话并复用临时表

import com.google.cloud.bigquery.*;

public class BigQuerySessionManager {

    public static void main(String[] args) throws InterruptedException {
        // 初始化 BigQuery 客户端
        BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();

        String sessionId = null;

        try {
            // 1. 创建临时表并初始化会话
            // 注意：临时表名需要以 _SESSION. 为前缀
            String createTempTableQuery = "CREATE TEMP TABLE _SESSION.my_temp_table AS " +
                                          "SELECT 1 AS id, 'Apple' AS fruit UNION ALL " +
                                          "SELECT 2, 'Banana' UNION ALL " +
                                          "SELECT 3, 'Cherry'";

            QueryJobConfiguration createSessionConfig =
                    QueryJobConfiguration.newBuilder(createTempTableQuery)
                            .setCreateSession(true) // 关键：设置为true以创建新会话
                            .build();

            // 提交作业并等待其完成
            Job createSessionJob = bigquery.create(JobInfo.of(createSessionConfig));
            createSessionJob = createSessionJob.waitFor(); // 等待作业完成

            if (createSessionJob.isDone() && createSessionJob.getStatus().getError() == null) {
                QueryStatistics queryStatistics = createSessionJob.getStatistics();
                if (queryStatistics != null && queryStatistics.getSessionInfo() != null) {
                    sessionId = queryStatistics.getSessionInfo().getSessionId();
                    System.out.println("会话创建成功。会话 ID: " + sessionId);
                } else {
                    System.err.println("未能从作业统计信息中获取会话信息。");
                }
            } else {
                System.err.println("创建会话作业失败: " + createSessionJob.getStatus().getError());
                // 如果作业失败，打印更多详细信息
                if (createSessionJob.getStatus().getExecutionErrors() != null) {
                    createSessionJob.getStatus().getExecutionErrors().forEach(System.err::println);
                }
            }

            if (sessionId == null) {
                System.err.println("未能获取会话 ID，程序终止。");
                return;
            }

            // 2. 复用会话进行后续查询 (将在下一节详细介绍)
            // ...

        } catch (BigQueryException e) {
            System.err.println("BigQuery 操作失败: " + e.getMessage());
            e.printStackTrace();
        }
    }
}

// ... (承接上一节代码，在获取到 sessionId 之后)

            // 2. 复用会话进行后续查询
            String selectFromTempTableQuery = "SELECT * FROM _SESSION.my_temp_table WHERE id = 1";
            QueryJobConfiguration reuseSessionConfig =
                    QueryJobConfiguration.newBuilder(selectFromTempTableQuery)
                            .setSessionId(sessionId) // 关键：使用之前获取的会话 ID
                            .build();

            // 提交查询作业并等待其完成
            Job reuseSessionJob = bigquery.create(JobInfo.of(reuseSessionConfig));
            reuseSessionJob = reuseSessionJob.waitFor(); // 等待作业完成

            if (reuseSessionJob.isDone() && reuseSessionJob.getStatus().getError() == null) {
                System.out.println("\n使用复用会话的查询成功完成。结果:");
                // 处理查询结果
                TableResult result = reuseSessionJob.getQueryResults();
                result.iterateAll().forEach(row -> {
                    System.out.println("ID: " + row.get("id").getStringValue() + ", Fruit: " + row.get("fruit").getStringValue());
                });
            } else {
                System.err.println("使用复用会话的查询失败: " + reuseSessionJob.getStatus().getError());
                if (reuseSessionJob.getStatus().getExecutionErrors() != null) {
                    reuseSessionJob.getStatus().getExecutionErrors().forEach(System.err::println);
                }
            }

            // 3. 在同一会话中执行另一个查询
            String anotherQuery = "SELECT COUNT(*) AS total_fruits FROM _SESSION.my_temp_table";
            QueryJobConfiguration anotherSessionConfig =
                    QueryJobConfiguration.newBuilder(anotherQuery)
                            .setSessionId(sessionId) // 再次复用会话 ID
                            .build();

            Job anotherJob = bigquery.create(JobInfo.of(anotherSessionConfig));
            anotherJob = anotherJob.waitFor();

            if (anotherJob.isDone() && anotherJob.getStatus().getError() == null) {
                System.out.println("\n同一会话中的另一个查询成功完成。结果:");
                TableResult result = anotherJob.getQueryResults();
                result.iterateAll().forEach(row -> {
                    System.out.println("总水果数: " + row.get("total_fruits").getStringValue());
                });
            } else {
                System.err.println("同一会话中的另一个查询失败: " + anotherJob.getStatus().getError());
                if (anotherJob.getStatus().getExecutionErrors() != null) {
                    anotherJob.getStatus().getExecutionErrors().forEach(System.err::println);
                }
            }

        } catch (BigQueryException e) {
            System.err.println("BigQuery 操作失败: " + e.getMessage());
            e.printStackTrace();
        }
    }
}