Feat/deltalake s3 #78

abduldjafar · 2024-12-06T23:35:28Z

What Kind of Change Does This PR Introduce?

This PR adds support for using DeltaLake with an S3 backend, expanding its capabilities beyond the current local filesystem option.

Current Behavior

DeltaLake only supports the local filesystem as the backend storage.

New Behavior

DeltaLake now supports S3 as a backend. Users can choose between the local filesystem and S3 for storing their DeltaLake data.

Configuration Instructions

To use DeltaLake with an S3 backend, ensure your AWS configuration is set up as shown in the example below:

If you are using a service like LocalStack, just add endpoint_url in config file like picture above

abduldjafar · 2024-12-10T22:31:36Z

hi @imor . for non API test i already test in local and its run well

imor · 2024-12-11T03:01:42Z

Hey @abduldjafar, I have committed a few fixes to your branch, please take a look. In particular the changes are:

Nit: fixed spacing and formatting in Cargo.toml and .vscode/settings.json files.
Dependencies in Cargo.toml are sorted alphabetically.
Removed maplit dependencies. We add dependencies only when absolutely needed because they add extra compilation and security burden. A crate like maplit doesn't add much value, we can just write the code by hand and it's not much worse.
Removed _write_to_table table as it wasn't called from anywhere.
get_lsn_schema function has been shortened.
Error messages start with small letters.
open_delta_table function was always discarding the error.

Please also double check the data type mappings in postgres_to_delta and postgres_to_arrow as some of them look wrong. E.g. in &Type::INT2 | &Type::INT4 | &Type::INT8 => ArrowDataType::Int32 all integer types are mapped to a 4 byte integer which for smaller types will waste space and for large types truncate the values.

Also in cell_to_arrow why are bytes converted to a string?

            Cell::Bytes(value) => {
                let data = std::str::from_utf8(value)
                    .map_err(|e| format!("Failed to convert to string: {}", e));
                let result = match data {
                    Ok(value) => value.to_string(),
                    Err(err_msg) => err_msg,
                };
                Arc::new(StringArray::from(vec![result]))
            }

There's no guarantee that the bytes type can be converted to a utf8 string. I think a thorough review of data types mapping is needed.

abduldjafar · 2024-12-11T03:57:08Z

Thanks a lot for the feedback, @imor! Regarding cell_to_arrow, I initially couldn't find the matching type, but let me double-check it again.

imor · 2024-12-11T04:07:14Z

Also could you please take a look at why the tests are failing.

abduldjafar · 2024-12-17T04:58:12Z

Hi @imor ready to review

pg_replicate/src/clients/delta.rs

imor · 2024-12-17T06:04:47Z

What was the reason for failing tests and how were they fixed?

abduldjafar · 2024-12-17T06:22:04Z

What was the reason for failing tests and how were they fixed?

Some sources said its because resources limitation. so the runner kill automatically. so in test part i add --job=1 . this will execute compiling process one by one. and will not consume much resource.

Co-authored-by: Raminder Singh <[email protected]>

imor · 2024-12-17T06:06:29Z

.github/workflows/general.yml

@@ -50,7 +50,7 @@ jobs:
      #     key: sqlx
      - name: Run non-api tests
        run: |
-          cargo test --workspace --all-features --exclude api
+          cargo test --workspace --all-features --exclude api --jobs 1


Why are jobs restricted to 1 here?

#78 (comment)

imor · 2024-12-17T06:12:37Z

pg_replicate/src/clients/delta.rs

+        match config.endpoint_url() {
+            Some(endpoint) => {
+                storage_options.insert(AWS_ENDPOINT_URL.to_string(), endpoint.to_string());
+                storage_options.insert(AWS_ALLOW_HTTP.to_string(), "true".to_string());
+            }
+            None => (),
+        }


Please run cargo clippy because clippy is emitting a warning in this code and suggesting to change it to an if let expression.

already run clippy and it passed in my local

imor · 2024-12-17T06:13:33Z

pg_replicate/src/clients/delta.rs

-                Arc::new(StringArray::from(vec![result]))
+                let data: Vec<&[u8]> = value
+                    .iter()
+                    .map(|item| std::slice::from_ref(item))


Another clippy warning.

imor · 2024-12-17T06:16:37Z

pg_replicate/src/clients/delta.rs

+        match config.endpoint_url() {
+            Some(endpoint) => {
+                storage_options.insert(AWS_ENDPOINT_URL.to_string(), endpoint.to_string());
+                storage_options.insert(AWS_ALLOW_HTTP.to_string(), "true".to_string());


Why do we want to enable insecure non-TLS connections?

this one if we want to test in minio or localstack

imor · 2024-12-17T06:18:33Z

pg_replicate/src/clients/delta.rs

+        let mut storage_options = HashMap::new();
+
+        storage_options.insert(AWS_FORCE_CREDENTIAL_LOAD.to_string(), "true".to_string());
+        storage_options.insert(AWS_S3_ALLOW_UNSAFE_RENAME.to_string(), "true".to_string());


Again, why no concurrent writer protection?

imor · 2024-12-17T06:27:30Z

pg_replicate/src/clients/delta.rs

+            &Type::INT8 => DataType::LONG,
+            &Type::INT2 => DataType::SHORT,
+            &Type::FLOAT4 => DataType::FLOAT,
+            &Type::FLOAT8 | &Type::NUMERIC => DataType::DOUBLE,


I don't think PG's Numeric type will fit in a double. It should be a decimal type.

do you know the precission?

imor · 2024-12-17T06:31:22Z

pg_replicate/src/clients/delta.rs

+            &Type::TIME | &Type::TIMESTAMP | &Type::TIMESTAMPTZ => {
+                DataType::Primitive(PrimitiveType::TimestampNtz)
+            }


Do all of three of these types map to PrimitiveType::TimestampNtz correctly?

imor · 2024-12-17T06:32:08Z

pg_replicate/src/clients/delta.rs

+            &Type::INT4 => ArrowDataType::Int32,
+            &Type::INT8 => ArrowDataType::Int64,
+            &Type::FLOAT4 => ArrowDataType::Float32,
+            &Type::FLOAT8 | &Type::NUMERIC => ArrowDataType::Float64,


Again, numeric shouldn't be mapped to float64

imor · 2024-12-17T06:37:18Z

pg_replicate/src/clients/delta.rs

+            .iter()
+            .map(|col| StructField::new(col.name.as_str(), Self::postgres_to_delta(&col.typ), true))
+            .chain([
+                StructField::new("OP", Self::postgres_to_delta(&Type::VARCHAR), true),


Why is this column nullable? Wouldn't its value be always present?

imor · 2024-12-17T06:37:47Z

pg_replicate/src/clients/delta.rs

+                StructField::new(
+                    "pg_replicate_inserted_time",
+                    Self::postgres_to_delta(&Type::TIMESTAMP),
+                    true,


Same for this column, why is it nullable?

imor · 2024-12-17T06:45:06Z

Also in the function cell_to_arrow, Cell::Null is mapped to an empty string, is this correct?

imor · 2024-12-17T06:48:19Z

pg_replicate/src/pipeline/sinks/delta.rs

+                    .create_table(&table_name, &table_schema.column_schemas)
+                    .await?
+            } else {
+                info!("Table already exists: {}", table_name);


Remove this?

Abdul Haris Djafar added 5 commits December 5, 2024 21:21

deltalake with aws backend

9b4b4de

deltalake with aws backend

39dc0fc

update code final update

a1f52a6

update code fix rustfmt issue

aeedc2a

update code add delta in vscode for consistency

c7dfd31

abduldjafar closed this Dec 10, 2024

abduldjafar reopened this Dec 10, 2024

imor added 8 commits December 11, 2024 06:53

fix formatting

9f1659b

reorder dependencies

4a45506

remove maplit and use String type as arg

346d1e4

refactor a couple of functions

5f840ac

simplify get_lsn_schema

e0742b6

improve error messages

3df23e8

remove unnecessary function

999141d

remove long prefix

c6acf00

Abdul Haris Djafar added 11 commits December 13, 2024 11:36

update code fix data type mapping

2ccd5f8

update code fix ci/cd

6a9c757

update code fix ci/cd

8a81a81

update code fix ci/cd

e883221

update code fix ci/cd

25461bf

update code fix ci/cd

ec6f4c6

update code fix ci/cd

c4208aa

update code fix test_non_api issues

eeb25e6

update code fix test_non_api issues

fae70dd

update code fix test_non_api issues

1e11bb3

update code fix test_non_api issues

089f52f

update code fix test_non_api issues

c52ef74

abduldjafar marked this pull request as draft December 15, 2024 03:39

update code fix s3 sink issue

703254d

abduldjafar marked this pull request as ready for review December 16, 2024 05:50

imor requested changes Dec 17, 2024

View reviewed changes

Update pg_replicate/src/clients/delta.rs

8556582

Co-authored-by: Raminder Singh <[email protected]>

imor reviewed Dec 17, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Feat/deltalake s3 #78

Feat/deltalake s3 #78

abduldjafar commented Dec 6, 2024 •

edited

Loading

abduldjafar commented Dec 10, 2024

imor commented Dec 11, 2024 •

edited

Loading

abduldjafar commented Dec 11, 2024

imor commented Dec 11, 2024

abduldjafar commented Dec 17, 2024

imor commented Dec 17, 2024

abduldjafar commented Dec 17, 2024

imor Dec 17, 2024

abduldjafar Dec 17, 2024

imor Dec 17, 2024

abduldjafar Dec 17, 2024

imor Dec 17, 2024

imor Dec 17, 2024

abduldjafar Dec 17, 2024

imor Dec 17, 2024

imor Dec 17, 2024

abduldjafar Dec 17, 2024

imor Dec 17, 2024

imor Dec 17, 2024

imor Dec 17, 2024

imor Dec 17, 2024

imor commented Dec 17, 2024

imor Dec 17, 2024

Feat/deltalake s3 #78

Are you sure you want to change the base?

Feat/deltalake s3 #78

Conversation

abduldjafar commented Dec 6, 2024 • edited Loading

What Kind of Change Does This PR Introduce?

Current Behavior

New Behavior

Configuration Instructions

abduldjafar commented Dec 10, 2024

imor commented Dec 11, 2024 • edited Loading

abduldjafar commented Dec 11, 2024

imor commented Dec 11, 2024

abduldjafar commented Dec 17, 2024

imor commented Dec 17, 2024

abduldjafar commented Dec 17, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

imor commented Dec 17, 2024

Choose a reason for hiding this comment

abduldjafar commented Dec 6, 2024 •

edited

Loading

imor commented Dec 11, 2024 •

edited

Loading