deploy pg service for Rust CI

Signed-off-by: zenghua <[email protected]>

.github/workflows/rust-ci.yml

@@ -24,6 +24,27 @@ env:
 jobs:
   rust_ci:
     runs-on: ubuntu-latest
+    services:
+      # Label used to access the service container
+      postgres:
+        # Docker Hub image
+        image: postgres:14.5
+        # Provide the password for postgres
+        env:
+          POSTGRES_PASSWORD: lakesoul_test
+          POSTGRES_USER: lakesoul_test
+          POSTGRES_DB: lakesoul_test
+        # Set health checks to wait until postgres has started
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+          --name lakesoul-test-pg
+        ports:
+          # Maps tcp port 5432 on service container to the host
+          - 5432:5432
+
     steps:
       - uses: actions/checkout@v3
       - uses: actions-rs/toolchain@v1

rust/lakesoul-metadata/src/lib.rs

@@ -14,6 +14,9 @@ pub use tokio::runtime::{Builder, Runtime};
 pub use tokio_postgres::{NoTls, Client, Statement};
 use postgres_types::{ToSql, FromSql};
 
+mod metadata_client;
+pub use metadata_client::MetaDataClient;
+
 pub const DAO_TYPE_QUERY_ONE_OFFSET : i32 = 0;
 pub const DAO_TYPE_QUERY_LIST_OFFSET : i32 = 100;
 pub const DAO_TYPE_INSERT_ONE_OFFSET : i32 = 200;
@@ -216,7 +219,7 @@ fn get_prepared_statement(
                 // Select PartitionInfo
                 DaoType::SelectPartitionVersionByTableIdAndDescAndVersion =>
                     "select table_id, partition_desc, version, commit_op, snapshot, expression, domain 
-                    from partition_info  
+                    from partition_info
                     where table_id = $1::TEXT and partition_desc = $2::TEXT and version = $3::INT",
                 DaoType::SelectOnePartitionVersionByTableIdAndDesc =>
                     "select m.table_id, t.partition_desc, m.version, m.commit_op, m.snapshot, m.expression, m.domain from (
@@ -1260,7 +1263,7 @@ pub fn execute_query_scalar(
             });
             match result {
                 Ok(Some(row)) => {
-                    let ts = row.get::<_, Option<i64>>(0);
+                    let ts = row.get::<_, Option<i32>>(0);
                     match ts {
                         Some(ts) => Ok(Some(format!("{}", ts))),
                         None => Ok(None)
@@ -1286,7 +1289,7 @@ pub fn execute_query_scalar(
             });
             match result {
                 Ok(Some(row)) => {
-                    let ts = row.get::<_, Option<i32>>(0);
+                    let ts = row.get::<_, Option<i64>>(0);
                     match ts {
                         Some(ts) => Ok(Some(format!("{}", ts))),
                         None => Ok(None)

rust/lakesoul-metadata/src/metadata_client.rs

@@ -19,24 +19,37 @@ pub struct MetaDataClient {
     prepared: PreparedStatementMap,
 }
 
-impl MetaDataClient {
-    pub fn from_env() -> Self{
-        let config_path = env::var("lakesoul_home").expect("lakesoul_home should be configured");
-        let config = fs::read_to_string(config_path).expect("");
-        let config_map = config.split('\n').filter_map(|property| {
-            property.find('=').map(|idx| property.split_at(idx + 1))
-        }).collect::<HashMap<_, _>>();
-        let url = Url::parse(&config_map.get("lakesoul.pg.url=").expect("")[5..]).unwrap();
+impl Default for MetaDataClient {
+    fn default() -> Self {
         Self::from_config(
-            format!(
-                "host={} port={} dbname={} user={} password={}", 
-                url.host_str().unwrap(), 
-                url.port().unwrap(), 
-                url.path_segments().unwrap().next().unwrap(),
-                config_map.get("lakesoul.pg.username=").unwrap(), 
-                config_map.get("lakesoul.pg.password=").unwrap())
+            "host=127.0.0.1 port=5432 dbname=lakesoul_test user=lakesoul_test password=lakesoul_test".to_string()
         )
     }
+}
+
+impl MetaDataClient {
+    pub fn from_env() -> Self{
+        match env::var("lakesoul_home") {
+            Ok(config_path) => {
+                let config = fs::read_to_string(&config_path).unwrap_or_else(|_| panic!("Fails at reading config file {}", &config_path));
+                let config_map = config.split('\n').filter_map(|property| {
+                    property.find('=').map(|idx| property.split_at(idx + 1))
+                }).collect::<HashMap<_, _>>();
+                let url = Url::parse(&config_map.get("lakesoul.pg.url=").unwrap_or(&"jdbc:postgresql://127.0.0.1:5432/lakesoul_test?stringtype=unspecified")[5..]).unwrap();
+                Self::from_config(
+                    format!(
+                        "host={} port={} dbname={} user={} password={}", 
+                        url.host_str().unwrap(), 
+                        url.port().unwrap(), 
+                        url.path_segments().unwrap().next().unwrap(),
+                        config_map.get("lakesoul.pg.username=").unwrap_or(&"lakesoul_test"), 
+                        config_map.get("lakesoul.pg.password=").unwrap_or(&"lakesoul_test"))
+                )
+            }
+            Err(_) => MetaDataClient::default()
+        }
+
+    }
 
     pub fn from_config(config: String) -> Self {
         let runtime = Builder::new_multi_thread()