Repository: roapi/roapi (language: python)

Issue:
Postgres support 
Would be fantastic to support reading from Postgres in the data layer.


Discussion / hints:
the conntector-x dependency we are using right now already has postgres support, so someone just need to enable that :D
I will be happy to give it a go. Is anyone already working on this?
No one is working on this at the moment, @alefminus please feel free to pick this up!
Well, I managed to reproduce your "Cannot start a runtime from within a runtime. This happens because a function..." error, so I guess this refers to the async runtime, which I admit I have little experience with. I guess I need to figure out how to make call the non block_on, i.e. the async version of something.. pointers welcome..
Can you paste the exact error stack trace here? I am curious what code path is creating a nested tokio runtime here.
It is the postgres connection.

```
at 23:38:06 ❯ cargo +nightly run --features database --bin columnq -- sql
-t ***@***.***/alon' 'select version()'
   Compiling columnq v0.5.0 (/home/alon/src/roapi/columnq)
   Compiling columnq-cli v0.3.0 (/home/alon/src/roapi/columnq-cli)
    Finished dev [unoptimized + debuginfo] target(s) in 10.62s
     Running `target/debug/columnq sql -t
***@***.***/alon'
'select version()'`
thread 'main' panicked at 'Cannot start a runtime from within a runtime.
This happens because a function (like `block_on`) attempted to block the
current thread while the thread is being used to drive asynchronous
tasks.',
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.21.1/src/runtime/scheduler/current_thread.rs:516
:26
note: run with `RUST_BACKTRACE=1` environment variable to display a
backtrace
thread 'main' panicked at 'Cannot start a runtime from within a runtime.
This happens because a function (like `block_on`) attempted to block the
current thread while the thread is being used to drive asynchronous
tasks.',
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/postgres-0.19.4/src/connection.rs:66
:22
stack backtrace:
   0:     0x557fc05472f0 -
std::backtrace_rs::backtrace::libunwind::trace::h1bc581f93c3c354e
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/../../backtrace/src/backtrace/mod.rs:66
:5
   1:     0x557fc05472f0 -
std::backtrace_rs::backtrace::trace_unsynchronized::h57190e335ed72b32
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/../../backtrace/src/backtrace/mod.rs:66
:5
   2:     0x557fc05472f0 -
std::sys_common::backtrace::_print_fmt::h99eb220d3e9328f4
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/sys_common/backtrace.rs:66
:5
   3:     0x557fc05472f0 -
<std::sys_common::backtrace::_print::DisplayBacktrace as
core::fmt::Display>::fmt::he73bd075427bad84
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/sys_common/backtrace.rs:45
:22
   4:     0x557fc0571e9e - core::fmt::write::hc5c99ee2b06a3be5
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/core/src/fmt/mod.rs:1202
:17
   5:     0x557fc0540665 - std::io::Write::write_fmt::hc372b845756ee1b5
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/io/mod.rs:1679
:15
   6:     0x557fc0548b23 -
std::sys_common::backtrace::_print::h1d3d2bb2d920f1a5
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/sys_common/backtrace.rs:48
:5
   7:     0x557fc0548b23 -
std::sys_common::backtrace::print::h11f80ce78ebf583b
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/sys_common/backtrace.rs:35
:9
   8:     0x557fc0548b23 -
std::panicking::default_hook::{{closure}}::h912e4ec2a3499b6f
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/panicking.rs:267
:22
   9:     0x557fc054880f - std::panicking::default_hook::h0e8b04fc1b26c86d
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/panicking.rs:286
:9
  10:     0x557fc0549229 -
std::panicking::rust_panic_with_hook::h81c35bb17ea74d5b
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/panicking.rs:669
:13
  11:     0x557fc050d948 -
std::panicking::begin_panic::{{closure}}::h63725f9e2dc4e472
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/panicking.rs:589
:9
  12:     0x557fc050e216 -
__rust_end_short_backtrace<std::panicking::begin_panic::{closure_env#0}<&str>,
!>
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/sys_common/backtrace.rs:138
:18
  13:     0x557fc050d880 - begin_panic<&str>
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/panicking.rs:588
:12
  14:     0x557fbf550a5d - tokio::runtime::enter::enter::h593839d898b2a9af
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.21.1/src/runtime/enter.rs:40
:9
  15:     0x557fbc8a265a -
block_on<futures_util::future::poll_fn::PollFn<postgres::connection::{impl#0}::poll_block_on::{closure_env#0}<postgres::client::{impl#1}::close_inner::{closure_env#0},
()>>>
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.21.1/src/runtime/scheduler/current_thread.rs:163
:33
  16:     0x557fbc8b7af9 -
tokio::runtime::Runtime::block_on::h955bc3f62ba010a7
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.21.1/src/runtime/mod.rs:490
:46
  17:     0x557fbc8b17a1 -
postgres::connection::Connection::poll_block_on::hd7cc5191539c5968
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/postgres-0.19.4/src/connection.rs:66
:9
  18:     0x557fbc8af356 -
postgres::client::Client::close_inner::hde8dc8b4b42d7192
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/postgres-0.19.4/src/client.rs:571
:9
  19:     0x557fbc8af18e - <postgres::client::Client as
core::ops::drop::Drop>::drop::h753f8a134bf2ae34
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/postgres-0.19.4/src/client.rs:20
:17
  20:     0x557fbc8b99d7 -
core::ptr::drop_in_place<postgres::client::Client>::hf220740e0472e056
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/core/src/ptr/mod.rs:487
:1
  21:     0x557fbc484227 -
core::ptr::drop_in_place<r2d2::Conn<postgres::client::Client>>::h3e7a4af38fa2ee40
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/core/src/ptr/mod.rs:487
:1
  22:     0x557fbc48c58b - r2d2::Pool<M>::try_get_inner::hb3b256c189722709
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/r2d2-0.8.10/src/lib.rs:483
:13
  23:     0x557fbc48ac4d - r2d2::Pool<M>::get_timeout::hb6a1ce6368213001
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/r2d2-0.8.10/src/lib.rs:424
:19
  24:     0x557fbc48d6b4 - r2d2::Pool<M>::get::h91f0f68352e674d8
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/r2d2-0.8.10/src/lib.rs:411
:9
  25:     0x557fbc550e28 -
<connectorx::sources::postgres::PostgresSource<P,C> as
connectorx::sources::Source>::fetch_metadata::h371bd48ff9eec39b
                               at
/home/alon/.cargo/git/checkouts/connector-x-da5b51094e647cf4/47e4061/connectorx/src/sources/postgres/mod.rs:151
:24
  26:     0x557fbc49cfb3 -
run<connectorx::sources::postgres::PostgresSource<connectorx::sources::postgres::BinaryProtocol,
tokio_postgres::tls::NoTls>,
connectorx::sources::postgres::typesystem::PostgresTypeSystem,
connectorx::destinations::arrow::ArrowDestination,
connectorx::destinations::arrow::typesystem::ArrowTypeSystem,
connectorx::transports::postgres_arrow::PostgresArrowTransport<connectorx::sources::postgres::BinaryProtocol,
tokio_postgres::tls::NoTls>,
connectorx::sources::postgres::errors::PostgresSourceError,
connectorx::destinations::arrow::errors::ArrowDestinationError,
connectorx::transports::postgres_arrow::PostgresArrowTransportError>
                               at
/home/alon/.cargo/git/checkouts/connector-x-da5b51094e647cf4/47e4061/connectorx/src/dispatcher.rs:62
:9
  27:     0x557fbbb6fd41 - to_mem_table
                               at
/home/alon/src/roapi/columnq/src/table/database.rs:94:25
  28:     0x557fbb9170a2 - {async_fn#0}
                               at
/home/alon/src/roapi/columnq/src/table/mod.rs:470:26
  29:     0x557fbb910a11 - poll<columnq::table::load::{async_fn_env#0}>
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/core/src/future/mod.rs:91
:19
  30:     0x557fbb7bbce0 - {async_fn#0}
                               at
/home/alon/src/roapi/columnq/src/columnq.rs:38:35
  31:     0x557fbb911681 -
poll<columnq::columnq::{impl#0}::load_table::{async_fn_env#0}>
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/core/src/future/mod.rs:91
:19
  32:     0x557fbb84d5e0 - {async_fn#0}
                               at
/home/alon/src/roapi/columnq-cli/src/main.rs:115:52
  33:     0x557fbb90fff1 - poll<columnq::cmd_sql::{async_fn_env#0}>
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/core/src/future/mod.rs:91
:19
  34:     0x557fbb84fc61 - {async_block#0}
                               at
/home/alon/src/roapi/columnq-cli/src/main.rs:209:67
  35:     0x557fbb90ecf1 - poll<columnq::main::{async_block_env#0}>
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/core/src/future/mod.rs:91
:19
  36:     0x557fbb842e70 -
{closure#0}<core::future::from_generator::GenFuture<columnq::main::{async_block_env#0}>>
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.21.1/src/park/thread.rs:267
:54
  37:     0x557fbb96d809 -
tokio::coop::with_budget::{{closure}}::h0c064537d2847215
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.21.1/src/coop.rs:102
:9
  38:     0x557fbb88e2b3 - try_with<core::cell::Cell<tokio::coop::Budget>,
tokio::coop::with_budget::{closure_env#0}<core::task::poll::Poll<core::result::Result<(),
anyhow::Error>>,
tokio::park::thread::{impl#5}::block_on::{closure_env#0}<core::future::from_generator::GenFuture<columnq::main::{async_block_env#0}>>>,
core::task::poll::Poll<core::result::Result<(), anyhow::Error>>>
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/thread/local.rs:445
:16
  39:     0x557fbb88da3d -
std::thread::local::LocalKey<T>::with::h2c9facb5647e3d7f
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/thread/local.rs:421
:9
  40:     0x557fbb842779 -
with_budget<core::task::poll::Poll<core::result::Result<(),
anyhow::Error>>,
tokio::park::thread::{impl#5}::block_on::{closure_env#0}<core::future::from_generator::GenFuture<columnq::main::{async_block_env#0}>>>
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.21.1/src/coop.rs:95
:5
  41:     0x557fbb842779 -
budget<core::task::poll::Poll<core::result::Result<(), anyhow::Error>>,
tokio::park::thread::{impl#5}::block_on::{closure_env#0}<core::future::from_generator::GenFuture<columnq::main::{async_block_env#0}>>>
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.21.1/src/coop.rs:72
:5
  42:     0x557fbb842779 -
block_on<core::future::from_generator::GenFuture<columnq::main::{async_block_env#0}>>
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.21.1/src/park/thread.rs:267
:31
  43:     0x557fbb802fb6 -
tokio::runtime::enter::Enter::block_on::h1fd602f6861edc1d
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.21.1/src/runtime/enter.rs:152
:13
  44:     0x557fbb96e709 -
tokio::runtime::scheduler::multi_thread::MultiThread::block_on::hcfdbc4b96b34fc5e
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.21.1/src/runtime/scheduler/multi_thread/mod.rs:79
:9
  45:     0x557fbb80e900 -
tokio::runtime::Runtime::block_on::h764cd387a0ac4cc7
                               at
/home/alon/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.21.1/src/runtime/mod.rs:492
:44
  46:     0x557fbb8150f9 - columnq::main::h98ddaa96d8e0d3c1
                               at
/home/alon/src/roapi/columnq-cli/src/main.rs:213:5
  47:     0x557fbb826a5b -
core::ops::function::FnOnce::call_once::h68fbea6e4b3e1225
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/core/src/ops/function.rs:251
:5
  48:     0x557fbb943cfe - __rust_begin_short_backtrace<fn() ->
core::result::Result<(), anyhow::Error>, core::result::Result<(),
anyhow::Error>>
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/sys_common/backtrace.rs:122
:18
  49:     0x557fbb83d0a1 - {closure#0}<core::result::Result<(),
anyhow::Error>>
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/rt.rs:166:18
  50:     0x557fc05391bf - core::ops::function::impls::<impl
core::ops::function::FnOnce<A> for &F>::call_once::hbaf9c3fc4d2fc8c8
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/core/src/ops/function.rs:286
:13
  51:     0x557fc05391bf - std::panicking::try::do_call::h675a505bd76bd09a
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/panicking.rs:464
:40
  52:     0x557fc05391bf - std::panicking::try::h6f2315ca797cf87a
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/panicking.rs:428
:19
  53:     0x557fc05391bf - std::panic::catch_unwind::hdd06926fcfca8628
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/panic.rs:137
:14
  54:     0x557fc05391bf -
std::rt::lang_start_internal::{{closure}}::hb49fec982844eb5c
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/rt.rs:148:48
  55:     0x557fc05391bf - std::panicking::try::do_call::hfa68f2d7caef5204
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/panicking.rs:464
:40
  56:     0x557fc05391bf - std::panicking::try::hfc3cce0df4217265
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/panicking.rs:428
:19
  57:     0x557fc05391bf - std::panic::catch_unwind::hf5baf9349e86ebcb
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/panic.rs:137
:14
  58:     0x557fc05391bf - std::rt::lang_start_internal::hfad0c5cf02bfe391
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/rt.rs:148:20
  59:     0x557fbb83d07a - std::rt::lang_start::hfd48eeffbbe90f80
                               at
/rustc/8b705839cd656d202e920efa8769cbe43a5ee269/library/std/src/rt.rs:165:17
  60:     0x557fbb8151b1 - main
  61:     0x7f57f3d31550 - __libc_start_call_main
  62:     0x7f57f3d31609 - __libc_start_main_impl
  63:     0x557fbb7ad365 - _start
  64:                0x0 - <unknown>
thread panicked while panicking. aborting.
Aborted (core dumped)

roapi on  postgres-support-177 [!?] via 🦀 v1.64.0 on ☁️  (us-east-1)
underwent 11s
```

The code change is: (also pushed to
https://github.com/alefminus/roapi/blob/postgres-support-177/columnq/src/table/database.rs#L78
)
```
                 DatabaseLoader::Postgres => {
-                    // ToDo `Cannot start a runtime from within a runtime`
error in `connector-x PostgresSource`
-                    return Err(ColumnQError::Database(
-                        "Postgres database features not be supported for
now.".to_string(),
-                    ));
+                    #[cfg(feature = "database-postgres")]
+                    {
+                        use
connectorx::sources::postgres::{BinaryProtocol, rewrite_tls_args};
+                        use tokio_postgres::NoTls;
+                        // TODO: avoid the second runtime ; should be able
to do everything without
+                        // invoking postgres methods, instead invoking
tokio_postgres directly (postgres
+                        // is a wrapper using block_on around postgres)
+                        use url::Url;
+                        let url = Url::parse(t.get_uri_str()).unwrap();
+                        let (config, _tls) =
rewrite_tls_args(&url).unwrap();
+                        let source = PostgresSource::<BinaryProtocol,
NoTls>::new(config, NoTls, 2)
+                            .map_err(|e|
ColumnQError::Database(e.to_string()))?;
+                        let dispatcher = Dispatcher::<
+                            PostgresSource<BinaryProtocol, NoTls>,
+                            ArrowDestination,
+                            PostgresArrowTransport<BinaryProtocol, NoTls>,
+                        >::new(
+                            source, &mut destination, queries, None
+                        );
+                        let mut main_runtime =
tokio::runtime::Runtime::new().unwrap();
+                        let cpu_pool =
tokio::runtime::Builder::new_multi_thread()
+                            .enable_all()
+                            .build()
+                            .unwrap();
+                        let cpu_pool = cpu_pool.handle().clone();
+
+                        main_runtime.spawn_blocking(|| {
+                            dispatcher
+                                .run()
+                                .map_err(|e|
ColumnQError::Database(e.to_string()))?;
+                        })1`
+                            .block_on(main_runtime.spawn(async move {
cpu_pool.spawn(async {}).await }))
+                            .unwrap()
+                            .unwrap();
+                        // run inside a second threaded runtime to avoid a
"Cannot start a runtime from within a runtime" error
+                        // because of postgres
+                    }
+                    #[cfg(not(feature = "database-postgresql"))]
+                    {
+                        // ToDo `Cannot start a runtime from within a
runtime` error in `connector-x PostgresSource`
+                        return Err(ColumnQError::Database(
+                            "Postgres database features not be supported
for now.".to_string(),
+                        ));
+                    }
                 }
```


On Thu, Sep 29, 2022 at 9:37 AM QP Hou ***@***.***> wrote:

> Can you paste the exact error stack trace here? I am curious what code
> path is creating a nested tokio runtime here.
>
> —
> Reply to this email directly, view it on GitHub
> <https://github.com/roapi/roapi/issues/177#issuecomment-1261832028>, or
> unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AYWFZPFSMD6LJWOGNQ2T2CDWAU2JJANCNFSM52RQFFZA>
> .
> You are receiving this because you were mentioned.Message ID:
> ***@***.***>
>


-- 
Alon Levy
Software Developer
Greenvibe
054-2395317

> Cannot start a runtime from within a runtime

looks very similar to https://github.com/sfu-db/connector-x/issues/218
Any plans when this will be fixed?
I don't think anyone is actively working on this right now, so PRs welcome @geoHeil :) The issue that @jychen7 referenced might be a reasonable workaround for now.

Fix patch:
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 7afdd89f3..32561c6bf 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -289,7 +289,7 @@ jobs:
         # set SDKROOT for C dependencies
         export SDKROOT=$(xcrun --sdk macosx --show-sdk-path)
         cd roapi && \
-          cargo build --bin roapi --features database --target aarch64-apple-darwin
+          cargo build --bin roapi --features database-sqlite,database-mysql  --target aarch64-apple-darwin
     - name: Trim cache
       run: |
         which cargo-cache || cargo install cargo-cache
diff --git a/Cargo.lock b/Cargo.lock
index 05ab66fd6..f4973e055 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1207,6 +1207,7 @@ dependencies = [
  "tempfile",
  "thiserror",
  "tokio",
+ "tokio-postgres",
  "toml 0.7.4",
  "uriparse",
  "url",
@@ -1271,24 +1272,34 @@ dependencies = [
  "anyhow",
  "arrow",
  "chrono",
+ "csv",
  "fallible-streaming-iterator",
  "fehler",
+ "hex",
  "itertools",
  "log",
  "mysql_common",
+ "native-tls",
  "num-traits",
+ "openssl",
  "owning_ref",
+ "postgres",
+ "postgres-native-tls",
+ "postgres-openssl",
  "r2d2",
  "r2d2_mysql",
+ "r2d2_postgres",
  "r2d2_sqlite",
  "rayon",
  "rusqlite",
  "rust_decimal",
+ "rust_decimal_macros",
  "serde_json",
  "sqlparser 0.11.0",
  "thiserror",
  "url",
  "urlencoding",
+ "uuid 0.8.2",
 ]
 
 [[package]]
@@ -1574,7 +1585,7 @@ dependencies = [
  "tokio-stream",
  "tokio-util",
  "url",
- "uuid",
+ "uuid 1.3.3",
  "xz2",
  "zstd 0.12.3+zstd.1.5.2",
 ]
@@ -1672,7 +1683,7 @@ dependencies = [
  "regex",
  "sha2 0.10.6",
  "unicode-segmentation",
- "uuid",
+ "uuid 1.3.3",
 ]
 
 [[package]]
@@ -1759,7 +1770,7 @@ dependencies = [
  "thiserror",
  "tokio",
  "url",
- "uuid",
+ "uuid 1.3.3",
 ]
 
 [[package]]
@@ -1865,7 +1876,7 @@ dependencies = [
  "rusoto_dynamodb",
  "thiserror",
  "tokio",
- "uuid",
+ "uuid 1.3.3",
 ]
 
 [[package]]
@@ -2998,7 +3009,7 @@ dependencies = [
  "subprocess",
  "thiserror",
  "time 0.3.22",
- "uuid",
+ "uuid 1.3.3",
 ]
 
 [[package]]
@@ -3517,6 +3528,32 @@ dependencies = [
  "tokio-postgres",
 ]
 
+[[package]]
+name = "postgres-native-tls"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2d442770e2b1e244bb5eb03b31c79b65bb2568f413b899eaba850fa945a65954"
+dependencies = [
+ "futures",
+ "native-tls",
+ "tokio",
+ "tokio-native-tls",
+ "tokio-postgres",
+]
+
+[[package]]
+name = "postgres-openssl"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1de0ea6504e07ca78355a6fb88ad0f36cafe9e696cbc6717f16a207f3a60be72"
+dependencies = [
+ "futures",
+ "openssl",
+ "tokio",
+ "tokio-openssl",
+ "tokio-postgres",
+]
+
 [[package]]
 name = "postgres-protocol"
 version = "0.6.5"
@@ -3542,8 +3579,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f028f05971fe20f512bcc679e2c10227e57809a3af86a7606304435bc8896cd6"
 dependencies = [
  "bytes",
+ "chrono",
  "fallible-iterator",
  "postgres-protocol",
+ "serde",
+ "serde_json",
+ "uuid 0.8.2",
 ]
 
 [[package]]
@@ -3681,6 +3722,16 @@ dependencies = [
  "r2d2",
 ]
 
+[[package]]
+name = "r2d2_postgres"
+version = "0.18.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7029c56be658cb54f321e0bee597810ee16796b735fa2559d7056bf06b12230b"
+dependencies = [
+ "postgres",
+ "r2d2",
+]
+
 [[package]]
 name = "r2d2_sqlite"
 version = "0.20.0"
@@ -3893,7 +3944,7 @@ dependencies = [
  "rkyv_derive",
  "seahash",
  "tinyvec",
- "uuid",
+ "uuid 1.3.3",
 ]
 
 [[package]]
@@ -4071,6 +4122,16 @@ dependencies = [
  "serde_json",
 ]
 
+[[package]]
+name = "rust_decimal_macros"
+version = "1.30.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7ca5c398d85f83b9a44de754a2048625a8c5eafcf070da7b8f116b685e2f6608"
+dependencies = [
+ "quote",
+ "rust_decimal",
+]
+
 [[package]]
 name = "rustc-hash"
 version = "1.1.0"
@@ -4844,6 +4905,18 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-openssl"
+version = "0.6.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c08f9ffb7809f1b20c1b398d92acf4cc719874b3b2b2d9ea2f09b4a80350878a"
+dependencies = [
+ "futures-util",
+ "openssl",
+ "openssl-sys",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-postgres"
 version = "0.7.8"
@@ -5170,6 +5243,12 @@ version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "711b9620af191e0cdc7468a8d14e709c3dcdb115b36f838e601583af800a370a"
 
+[[package]]
+name = "uuid"
+version = "0.8.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bc5cf98d8186244414c848017f0e2676b3fcb46807f6668a97dfe67359a3c4b7"
+
 [[package]]
 name = "uuid"
 version = "1.3.3"
diff --git a/README.md b/README.md
index f48a526c0..9903a0a3b 100644
--- a/README.md
+++ b/README.md
@@ -310,7 +310,7 @@ Data layer:
 - [x] Google spreadsheet
 - [x] MySQL
 - [x] SQLite
-- [ ] Postgres
+- [x] Postgres
 - [ ] Airtable
 - Data format
   - [x] CSV
diff --git a/columnq/Cargo.toml b/columnq/Cargo.toml
index 1e0502732..edee6fab5 100644
--- a/columnq/Cargo.toml
+++ b/columnq/Cargo.toml
@@ -13,7 +13,7 @@ path = "src/lib.rs"
 [dependencies]
 # pulling arrow-schema manually to enable the serde feature.
 # TODO: add serde feature in datafusion to avoid this workaround
-arrow-schema = { version ="37.0.0", features = ["serde"] }
+arrow-schema = { version = "37.0.0", features = ["serde"] }
 
 datafusion = "23"
 object_store = { version = "0.5.6", features = ["aws_profile", "gcp", "azure"] }
@@ -24,7 +24,7 @@ log = "0"
 regex = "1"
 lazy_static = "1"
 graphql-parser = "0"
-sqlparser = "0.33"  # version need to be in sync with convergence
+sqlparser = "0.33" # version need to be in sync with convergence
 yup-oauth2 = { version = "6.2", default-features = false, features = [
     "service_account",
 ] }
@@ -44,6 +44,7 @@ tokio = { version = "1", features = ["rt-multi-thread"] }
 futures = "0.3"
 hyper-tls = { version = "0.5.0", default-features = false, optional = true }
 hyper-rustls = { version = "0.23.2", default-features = false, optional = true }
+tokio-postgres = { version = "0.7.8", optional = true }
 
 [dependencies.deltalake]
 default-features = false
@@ -92,7 +93,9 @@ native-tls = [
 simd = ["datafusion/simd"]
 database-sqlite = ["connectorx/src_sqlite"]
 database-mysql = ["connectorx/src_mysql"]
+database-postgres = ["connectorx/src_postgres", "dep:tokio-postgres"]
 database = [
     "database-sqlite",
     "database-mysql",
+    "database-postgres"
 ]
diff --git a/columnq/src/table/database.rs b/columnq/src/table/database.rs
index 7f34c4bca..5fbe6a666 100644
--- a/columnq/src/table/database.rs
+++ b/columnq/src/table/database.rs
@@ -4,13 +4,19 @@ pub enum DatabaseLoader {
     Postgres,
 }
 
-#[cfg(any(feature = "database-sqlite", feature = "database-mysql"))]
+#[cfg(any(
+    feature = "database-sqlite",
+    feature = "database-mysql",
+    feature = "database-postgres"
+))]
 mod imp {
     use crate::error::ColumnQError;
     use crate::table::TableSource;
     use connectorx::prelude::*;
-    #[cfg(feature = "database-mysql")]
-    use connectorx::sources::mysql::BinaryProtocol;
+    #[cfg(any(feature = "database-mysql"))]
+    use connectorx::sources::mysql;
+    #[cfg(any(feature = "database-postgres"))]
+    use connectorx::sources::postgres;
     use datafusion::arrow::record_batch::RecordBatch;
     use log::debug;
 
@@ -28,18 +34,24 @@ mod imp {
                 DatabaseLoader::MySQL => {
                     #[cfg(feature = "database-mysql")]
                     {
-                        let source = MySQLSource::<BinaryProtocol>::new(t.get_uri_str(), 2)
+                        let source = MySQLSource::<mysql::BinaryProtocol>::new(t.get_uri_str(), 2)
                             .map_err(|e| ColumnQError::Database(e.to_string()))?;
                         let dispatcher = Dispatcher::<
-                            MySQLSource<BinaryProtocol>,
+                            MySQLSource<mysql::BinaryProtocol>,
                             ArrowDestination,
-                            MySQLArrowTransport<BinaryProtocol>,
+                            MySQLArrowTransport<mysql::BinaryProtocol>,
                         >::new(
                             source, &mut destination, queries, None
                         );
                         dispatcher
                             .run()
                             .map_err(|e| ColumnQError::Database(e.to_string()))?;
+                        let schema_ref = destination.arrow_schema();
+                        let data: Vec<RecordBatch> = destination.arrow().unwrap();
+                        Ok(datafusion::datasource::MemTable::try_new(
+                            schema_ref,
+                            vec![data],
+                        )?)
                     }
                     #[cfg(not(feature = "database-mysql"))]
                     {
@@ -64,6 +76,13 @@ mod imp {
                         dispatcher
                             .run()
                             .map_err(|e| ColumnQError::Database(e.to_string()))?;
+
+                        let schema_ref = destination.arrow_schema();
+                        let data: Vec<RecordBatch> = destination.arrow().unwrap();
+                        Ok(datafusion::datasource::MemTable::try_new(
+                            schema_ref,
+                            vec![data],
+                        )?)
                     }
                     #[cfg(not(feature = "database-sqlite"))]
                     {
@@ -73,23 +92,70 @@ mod imp {
                     }
                 }
                 DatabaseLoader::Postgres => {
-                    // ToDo `Cannot start a runtime from within a runtime` error in `connector-x PostgresSource`
-                    return Err(ColumnQError::Database(
-                        "Postgres database features not be supported for now.".to_string(),
-                    ));
+                    #[cfg(feature = "database-postgres")]
+                    {
+                        let config = t
+                            .get_uri_str()
+                            .parse::<tokio_postgres::Config>()
+                            .map_err(|e| ColumnQError::Database(e.to_string()))?;
+                        let tls = match config.get_ssl_mode() {
+                            tokio_postgres::config::SslMode::Require => tokio_postgres::NoTls,
+                            _ => tokio_postgres::NoTls,
+                        };
+                        let source: PostgresSource<
+                            postgres::BinaryProtocol,
+                            tokio_postgres::NoTls,
+                        > = PostgresSource::new(config.into(), tls, 2)
+                            .map_err(|e| ColumnQError::Database(e.to_string()))?;
+                        let queries = queries.clone();
+                        let task = tokio::task::spawn_blocking(move || {
+                            let dispatcher = Dispatcher::<
+                                PostgresSource<postgres::BinaryProtocol, tokio_postgres::NoTls>,
+                                ArrowDestination,
+                                PostgresArrowTransport<
+                                    postgres::BinaryProtocol,
+                                    tokio_postgres::NoTls,
+                                >,
+                            >::new(
+                                source, &mut destination, &queries, None
+                            );
+
+                            if let Err(e) = dispatcher.run() {
+                                return Err(ColumnQError::Database(e.to_string()));
+                            }
+
+                            let schema_ref = destination.arrow_schema();
+                            match destination.arrow() {
+                                Ok(data) => datafusion::datasource::MemTable::try_new(
+                                    schema_ref,
+                                    vec![data],
+                                )
+                                .map_err(|e| ColumnQError::Database(e.to_string())),
+                                Err(e) => Err(ColumnQError::Database(e.to_string())),
+                            }
+                        });
+
+                        // FIXME: Maybe use other way to block the async task instead of block_on
+                        futures::executor::block_on(task)
+                            .map_err(|e| ColumnQError::Database(e.to_string()))?
+                    }
+                    #[cfg(not(feature = "database-postgres"))]
+                    {
+                        return Err(ColumnQError::Database(
+                            "Postgres database feature not enabled.".to_string(),
+                        ));
+                    }
                 }
-            };
-            let schema_ref = destination.arrow_schema();
-            let data: Vec<RecordBatch> = destination.arrow().unwrap();
-            Ok(datafusion::datasource::MemTable::try_new(
-                schema_ref,
-                vec![data],
-            )?)
+            }
         }
     }
 }
 
-#[cfg(not(any(feature = "database-sqlite", feature = "database-mysql")))]
+#[cfg(not(any(
+    feature = "database-sqlite",
+    feature = "database-mysql",
+    feature = "database-postgres"
+)))]
 mod imp {
     use crate::error::ColumnQError;
     use crate::table::TableSource;


Test patch:
diff --git a/columnq/tests/table_postgres_test.rs b/columnq/tests/table_postgres_test.rs
new file mode 100644
index 000000000..357b5a611
--- /dev/null
+++ b/columnq/tests/table_postgres_test.rs
@@ -0,0 +1,24 @@
+#[cfg(any(feature = "database-postgres"))]
+mod postgres {
+    use datafusion::datasource::TableProvider;
+    use datafusion::prelude::SessionContext;
+    use std::env;
+
+    use columnq::table::TableSource;
+
+    use columnq::table::database::DatabaseLoader;
+
+    #[tokio::test]
+    async fn load_postgres() -> anyhow::Result<()> {
+        dotenvy::dotenv().ok();
+        if let Ok(name) = env::var("TABLE_NAME") {
+            let t = DatabaseLoader::Postgres
+                .to_mem_table(&TableSource::new(name, env::var("POSGRES_URL")?))?;
+            let ctx = SessionContext::new();
+            let stats = t.scan(&ctx.state(), None, &[], None).await?.statistics();
+            assert!(stats.num_rows.is_some());
+        }
+
+        Ok(())
+    }
+}