jonahgao commented on code in PR #13936:
URL: https://github.com/apache/datafusion/pull/13936#discussion_r1900008339
##########
datafusion/sqllogictest/bin/sqllogictests.rs:
##########
@@ -452,3 +758,87 @@ impl Options {
}
}
}
+
+#[cfg(feature = "postgres")]
+pub async fn start_postgres(
+ in_channel: &Channel<ContainerCommands>,
+ host_channel: &Channel<String>,
+ port_channel: &Channel<u16>,
+ stopped_channel: &Channel<()>,
+) {
+ info!("Starting postgres test container with user postgres/postgres and db
test");
+
+ let container = testcontainers_modules::postgres::Postgres::default()
+ .with_user("postgres")
+ .with_password("postgres")
+ .with_db_name("test")
+ .with_mapped_port(16432, 5432.tcp())
+ .with_tag("17-alpine")
+ .start()
+ .await
+ .unwrap();
+ // uncomment this if you are running docker in docker
+ // let host = "host.docker.internal".to_string();
+ let host = container.get_host().await.unwrap().to_string();
+ let port = container.get_host_port_ipv4(5432).await.unwrap();
+
+ let mut rx = in_channel.rx.lock().await;
+ while let Some(command) = rx.recv().await {
+ match command {
+ FetchHost => host_channel.tx.send(host.clone()).unwrap(),
+ FetchPort => port_channel.tx.send(port).unwrap(),
+ ContainerCommands::Stop => {
+ container.stop().await.unwrap();
+ stopped_channel.tx.send(()).unwrap();
+ rx.close();
+ }
+ }
+ }
+}
+
+#[cfg(feature = "postgres")]
+#[derive(Debug)]
+pub enum ContainerCommands {
+ FetchHost,
+ FetchPort,
+ Stop,
+}
+
+#[cfg(feature = "postgres")]
+pub struct Channel<T> {
+ pub tx: UnboundedSender<T>,
+ pub rx: Mutex<UnboundedReceiver<T>>,
+}
+
+#[cfg(feature = "postgres")]
+pub fn channel<T>() -> Channel<T> {
+ let (tx, rx) = mpsc::unbounded_channel();
+ Channel {
+ tx,
+ rx: Mutex::new(rx),
+ }
+}
+
+#[cfg(feature = "postgres")]
+pub fn execute_blocking<F: Future>(f: F) {
+ tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap()
+ .block_on(f);
+}
+
+#[cfg(feature = "postgres")]
+pub struct HostPort {
+ pub host: String,
+ pub port: u16,
+}
+
+#[cfg(feature = "postgres")]
+static POSTGRES_IN: Lazy<Channel<ContainerCommands>> = Lazy::new(channel);
Review Comment:
We can use `std::sync::LazyLock` without introducing the dependency of
`once_cell`.
##########
datafusion/sqllogictest/bin/sqllogictests.rs:
##########
@@ -97,23 +169,89 @@ async fn run_tests() -> Result<()> {
}
options.warn_on_ignored();
+ #[cfg(feature = "postgres")]
+ let start_pg_database = options.postgres_runner && !is_pg_uri_set();
+ #[cfg(feature = "postgres")]
+ if start_pg_database {
+ info!("Starting postgres db ...");
+
+ thread::spawn(|| {
+ execute_blocking(start_postgres(
+ &POSTGRES_IN,
+ &POSTGRES_HOST,
+ &POSTGRES_PORT,
+ &POSTGRES_STOPPED,
+ ))
+ });
+
+ POSTGRES_IN.tx.send(FetchHost).unwrap();
+ let db_host = POSTGRES_HOST.rx.lock().await.recv().await.unwrap();
Review Comment:
Here we need to wait for Postgres to start, so I wonder if we can call
`start_postgres` directly in the current thread, without using `thread::spawn`
and channels.
##########
datafusion/sqllogictest/src/engines/postgres_engine/mod.rs:
##########
@@ -194,16 +214,13 @@ fn no_quotes(t: &str) -> &str {
/// return a schema name
fn schema_name(relative_path: &Path) -> String {
relative_path
- .file_name()
- .map(|name| {
- name.to_string_lossy()
- .chars()
- .filter(|ch| ch.is_ascii_alphabetic())
- .collect::<String>()
- .trim_start_matches("pg_")
- .to_string()
- })
- .unwrap_or_else(|| "default_schema".to_string())
+ .to_string_lossy()
+ .to_string()
Review Comment:
nit: Calling `to_string()` seems unnecessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org
For additional commands, e-mail: github-h...@datafusion.apache.org
