FWD » Core Development » Database

import com.mchange.v2.c3p0.ComboPooledDataSource;

import java.sql.*;

import java.util.*;

import java.util.concurrent.CompletableFuture;

public class Test {

    private static final ComboPooledDataSource dataSource = new ComboPooledDataSource();

    static {

        try {

            dataSource.setDriverClass("org.postgresql.Driver");

            dataSource.setJdbcUrl("jdbc:postgresql://localhost:5433/fwd");

            dataSource.setUser("fwd_admin");

            dataSource.setPassword("admin");

            dataSource.setMinPoolSize(10);

            dataSource.setMaxPoolSize(100);

        } catch (Exception e) {

            throw new RuntimeException("Error setting up datasource", e);

        }

    }

    public static void executeParallelQueries() {

        CompletableFuture<List<Map<String, Object>>> future1 = CompletableFuture.supplyAsync(() ->

                runStreamingQuery("SELECT *, pg_sleep(0.001) FROM employees e WHERE e.id = 7001;")

        );

        CompletableFuture<List<Map<String, Object>>> future4 = CompletableFuture.supplyAsync(() ->

                runStreamingQuery("SELECT *, pg_sleep(0.001) FROM employees e WHERE e.id = 7001;")

        );

        CompletableFuture<List<Map<String, Object>>> future2 = CompletableFuture.supplyAsync(() ->

                runStreamingQuery("SELECT *, pg_sleep(0.001) FROM departments d WHERE d.id = 343")

        );

        CompletableFuture<List<Map<String, Object>>> future3 = CompletableFuture.supplyAsync(() ->

                runStreamingQuery("SELECT *, pg_sleep(0.001) FROM departments d WHERE d.id = 343")

        );

        // Wait and combine results

        CompletableFuture.allOf(future1, future2, future3, future4).join();

        try {

            List<Map<String, Object>> employees = future1.get();

            List<Map<String, Object>> departments = future2.get();

            List<Map<String, Object>> departments2 = future3.get();

            List<Map<String, Object>> employees2 = future4.get();

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

    private static List<Map<String, Object>> runStreamingQuery(String query) {

        List<Map<String, Object>> results = new ArrayList<>();

        try (Connection conn = dataSource.getConnection()) {

            conn.setAutoCommit(false);

            try (Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {

                stmt.setFetchSize(50);

                try (ResultSet rs = stmt.executeQuery(query)) {

                    ResultSetMetaData meta = rs.getMetaData();

                    int columnCount = meta.getColumnCount();

                    while (rs.next()) {

                        Map<String, Object> row = new HashMap<>();

                        for (int i = 1; i <= columnCount; i++) {

                            row.put(meta.getColumnLabel(i), rs.getObject(i));

                        }

                        results.add(row);

                    }

                }

            }

        } catch (SQLException e) {

            System.err.println("Error executing query: " + query);

            e.printStackTrace();

        }

        return results;

    }

    public static void main(String[] args) {

        double async = System.currentTimeMillis();

        for (int i = 0; i < 500; i++) {

            executeParallelQueries();

        }

        async = System.currentTimeMillis() - async;

        System.out.println(async);

        double sync = System.currentTimeMillis();

        for (int i = 0; i < 500; i++) {

            executeQueriesSequentially();

        }

        sync = System.currentTimeMillis() - sync;

        System.out.println(sync);

        System.out.println((sync - async) / sync * 100);

    }

    public static void executeQueriesSequentially() {

        List<Map<String, Object>> employees = runQueryNoStreaming("SELECT *, pg_sleep(0.001) FROM employees e WHERE e.id = 7001;");

        List<Map<String, Object>> employees2 = runQueryNoStreaming("SELECT *, pg_sleep(0.001) FROM employees e WHERE e.id = 7001;");

        List<Map<String, Object>> departments = runQueryNoStreaming("SELECT *, pg_sleep(0.001) FROM departments d WHERE d.id = 343");

        List<Map<String, Object>> departments2 = runQueryNoStreaming("SELECT *, pg_sleep(0.001) FROM departments d WHERE d.id = 343");

    }

    private static List<Map<String, Object>> runQueryNoStreaming(String query) {

        List<Map<String, Object>> results = new ArrayList<>();

        try (Connection conn = dataSource.getConnection()) {

            conn.setAutoCommit(true); // default auto commit on, no streaming

            try (Statement stmt = conn.createStatement()) {

                // No fetch size set, so all rows fetched eagerly

                try (ResultSet rs = stmt.executeQuery(query)) {

                    ResultSetMetaData meta = rs.getMetaData();

                    int columnCount = meta.getColumnCount();

                    while (rs.next()) {

                        Map<String, Object> row = new HashMap<>();

                        for (int i = 1; i <= columnCount; i++) {

                            row.put(meta.getColumnLabel(i), rs.getObject(i));

                        }

                        results.add(row);

                    }

                }

            }

        } catch (SQLException e) {

            System.err.println("Error executing query: " + query);

            e.printStackTrace();

        }

        return results;

    }

}