feat(sink): support deltalake sink with rust sdk #13600

xxhZs · 2023-11-22T11:19:29Z

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

support deltalake sink with rust sdk.

resolve #13152

Checklist

I have written necessary rustdoc comments
I have added necessary unit tests and integration tests
I have added test labels as necessary. See details.
I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
All checks passed in ./risedev check (or alias, ./risedev c)
My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

fmt fmt

xxchan · 2023-11-23T06:49:27Z

src/common/src/array/arrow_copy_with_arrow46.rs

We added arrow_copy_with_arrow46.rs to resolve conflict issues.

This doesn't look very cool. Why not upgrade deltalake's arrow version instead? (Can use a fork if upstream won't accept the upgrade and release a version soon)

Okk. I will make efforts to unify the versions of both.

modify cargo lock fix

wangrunji0408 · 2023-11-24T03:36:47Z

Cargo.toml

+arrow-select = "48.0.1"
+arrow-ord = "48.0.1"
+arrow-row = "48.0.1"
+deltalake = { git = "https://github.com/delta-io/delta-rs", rev = "72505449e9538371fe5fda35d545dbd662facd07", features = ["s3"] }


Is there a particular reason for downgrading arrow?

Because both Delta.rs and IceLake need to use Arrow, version 48.0.1 is the maximum version they both support.

wenym1

Rest LGTM.

src/connector/src/sink/deltalake.rs

wenym1 · 2023-11-27T06:18:17Z

src/connector/src/sink/deltalake.rs

+    }
+}
+
+fn covert_schema(schema: &StructType) -> Result<deltalake::arrow::datatypes::Schema> {


topo. covert -> convert.

This method looks like a general utils method. Is this method implemented in the delta lake sdk?

We only found the conversion method for datatype, so we need to write a loop iteration ourselves.

I just found that we can use deltalake::arrow::datatypes::Schema::try_from(dl_table.get_schema()?) in where we currently call convert_schema.

src/connector/src/sink/deltalake.rs

wenym1

Can we include some e2e test in e2e_test/?

src/connector/src/sink/deltalake.rs

wenym1 · 2023-11-30T10:31:10Z

Cargo.toml

-arrow-select = "49"
-arrow-ord = "49"
-arrow-row = "49"
+arrow-array = "48.0.1"


What compile error will we get if we don't use the arrow version of delta.rs? Does the compile error comes from unable to use the to_record_batch_with_schema implemented by us?

Yes. If we use 49. Cargo will reload a 48 on the dependent link for delta, and then the output of our to_record_batch_with_schema method is 49. It cannot be input into the write of delta because the parameter accepted by write is 48.
So it is necessary to ensure that the versions of 'arrow' relied upon by 'icelake' and 'delta' are the same.

See #13600 (comment).

wenym1 · 2023-12-01T03:59:03Z

In a new commit c890167, I just tried to introduce a separate arrow version for deltalake so that the arrow version of the current code will not be downgraded. This is achieved by manually setting the arrow version of icelake in Cargo.lock.

The current code of arrow datachunk conversion is moved to a macro gen_code, so that the conversion code can be applied to both versions.

The reason for delta-rs to depend on a older version (i.e. 48.0.1) of arrow is that its version of datafusion it depends on is 48.0.1. However, the latest version of datafusion is already using 49.0.0. So after the delta-rs upgrades their datafusion version to one that uses 49.0.0, we can then let both delta-rs to use the same version as our current code. FYI, I tried forked delta-rs and upgrades their datafusion version, but it cannot compile without some extra effort because there is some breaking changes in the upgrade. We can have a issue to track their version upgrade @xxhZs

@wangrunji0408 @xxchan Could you take a look whether the current hack looks good to you?

xxhZs · 2023-12-01T04:21:52Z

In a new commit c890167, I just tried to introduce a separate arrow version for deltalake so that the arrow version of the current code will not be downgraded. This is achieved by manually setting the arrow version of icelake in Cargo.lock.

The current code of arrow datachunk conversion is moved to a macro gen_code, so that the conversion code can be applied to both versions.

The reason for delta-rs to depend on a older version (i.e. 48.0.1) of arrow is that its version of datafusion it depends on is 48.0.1. However, the latest version of datafusion is already using 49.0.0. So after the delta-rs upgrades their datafusion version to one that uses 49.0.0, we can then let both delta-rs to use the same version as our current code. FYI, I tried forked delta-rs and upgrades their datafusion version, but it cannot compile without some extra effort because there is some breaking changes in the upgrade. We can have a issue to track their version upgrade @xxhZs

@wangrunji0408 @xxchan Could you take a look whether the current hack looks good to you?

Using macros to reuse two copies of code is acceptable, and we may need to keep it that way all the time. Because there will always be a time when icelack and delta.rs depend on an inconsistent arrow.
If we want to finally unify both of their dependencies on arrow, i.e. only one version of arrow exists in our system, I think we can totally use 48 in this pr, since there seems to be no component in our system that needs to work under 49, and after that we just need to wait for both to depend on one version (in fact, just need delta rs to upgrade to it, since icelake's dependency is >48, and the newer version is supported by default), and then update the arrow version.

wenym1 · 2023-12-01T04:33:14Z

Using macros to reuse two copies of code is acceptable, and we may need to keep it that way all the time. Because there will always be a time when icelack and delta.rs depend on an inconsistent arrow.
If we want to finally unify both of their dependencies on arrow, i.e. only one version of arrow exists in our system, I think we can totally use 48 in this pr, since there seems to be no component in our system that needs to work under 49, and after that we just need to wait for both to depend on one version (in fact, just need delta rs to upgrade to it, since icelake's dependency is >48, and the newer version is supported by default), and then update the arrow version.

Not only deltalake and icelake are using arrow, but also our internal udf. In the future if we upgrade the arrow version of udf, it doesn't make sense to be blocked by arrow version of some external connectors. I think it's acceptable to support several versions of arrow at the same time if we don't have to duplicate the code.

github-actions

license-eye has totally checked 4509 files.

Valid	Invalid	Ignored	Fixed
1982	3	2524	0

Click to see the invalid file list

src/common/src/array/arrow/arrow_common.rs
src/common/src/array/arrow/arrow_deltalake.rs
src/common/src/array/arrow/mod.rs

src/common/src/array/arrow/arrow_common.rs

src/common/src/array/arrow/arrow_deltalake.rs

src/common/src/array/arrow/mod.rs

wenym1 · 2023-12-01T05:40:17Z

I just changed to reload the same arrow.rs for twice for different arrow version. Now we don't even need to use macro.

pub use arrow_impl::to_record_batch_with_schema as to_deltalake_record_batch_with_schema;
use {
    arrow_array_deltalake as arrow_array, arrow_buffer_deltalake as arrow_buffer,
    arrow_cast_deltalake as arrow_cast, arrow_schema_deltalake as arrow_schema,
};

#[allow(clippy::duplicate_mod)]
#[path = "./arrow.rs"]
mod arrow_impl;

In the outer mod we rename the arrow packages to arrow_xxx, and in arrow.rs we use super::arrow_xxx to use arrow version specified in the outer mod.

e2e_test/sink/deltalake_rust_sink.slt

fix ci fix ci fix fix ci fix ci fix

Cargo.toml

src/connector/Cargo.toml

src/connector/src/sink/deltalake.rs

wenym1

LGTM! Thanks for the PR.

Unit test time out in CI because we add new test case in this PR. we may increase the timeout a bit to pass the CI.

e2e_test/sink/deltalake_rust_sink.slt

hzxa21

There seems to be so many Cargo.lock changes. Is this expected? Also, will this PR increase the compile time and binary size significantly given that many new dependencies are added. cc @xxhZs @wenym1

wenym1 · 2023-12-13T02:33:29Z

There seems to be so many Cargo.lock changes. Is this expected? Also, will this PR increase the compile time and binary size significantly given that many new dependencies are added.

This is expected because most changes are introduced by the delta lake libraries.

I had a test locally on the main branch and this branch to test building the risingwave_cmd_all in release mode.

	# of deps	compile time	binary size
current commit	1222	19m 36s	381,807,328
current commit w/o datafusion	1204	16m 47s	340898016
current commit w/o datafusion + dynamo db	1178	16m 20s	332395424
main	1148	16m 05s	316,990,384
main + arrow 0.48	1154	16m 32s	317,187,120

It looks like the great number of changes in Cargo.lock are introduced by datafusion and dynamo db. The datafusion is introduced by accidence in the dev-dependency and can be temporarily removed. The dynamo db dependency is bundled in the delta rs sdk when we enable the s3 feature of it. I have forked the repo and support conditionally removing the dependency on dynamo db so that we can avoid introducing the dynamo db sdk dependency.

After removing the dependency on datafusion and dynamo db, the compile time and binary size is greatly reduced, which is slightly increases compared to the main.

Besides, I had a wired observation that the datafusion dependency is introduced in dev-dependency and only used in unit tests. However, the Cargo.lock still include datafusion as a dependency of risingave_connector, which causes the increase in compile time.

xxhZs added 2 commits November 21, 2023 12:15

add delta lake sink

e6bc0ee

fix

013eede

fmt fmt

xxhZs linked an issue Nov 22, 2023 that may be closed by this pull request

Reimplement delta lake sink with rust sdk #13152

Closed

xxhZs added 2 commits November 23, 2023 14:31

Merge branch 'main' into 13152-reimplement-delta-lake-sink-with-rust-sdk

f0047ba

fmt

2ff021f

xxhZs marked this pull request as ready for review November 23, 2023 06:34

xxhZs requested a review from a team as a code owner November 23, 2023 06:34

xxchan reviewed Nov 23, 2023

View reviewed changes

xxhZs added 2 commits November 24, 2023 11:13

save

7b85ea2

modify cargo lock fix

fix cargo lock

6fbbac3

github-actions bot added the type/feature label Nov 24, 2023

xxhZs requested review from hzxa21 and wenym1 November 24, 2023 03:28

wangrunji0408 reviewed Nov 24, 2023

View reviewed changes

wenym1 reviewed Nov 27, 2023

View reviewed changes

xxhZs and others added 4 commits November 28, 2023 13:22

fmt

c6a709c

fix

1ddf1f9

Merge branch 'main' into 13152-reimplement-delta-lake-sink-with-rust-sdk

cdecda5

Merge branch 'main' into 13152-reimplement-delta-lake-sink-with-rust-sdk

f521372

xiangjinwu reviewed Nov 29, 2023

View reviewed changes

src/connector/src/sink/deltalake.rs Outdated Show resolved Hide resolved

wenym1 reviewed Nov 30, 2023

View reviewed changes

use separate arrow version for deltalake

c890167

Merge branch 'main' into 13152-reimplement-delta-lake-sink-with-rust-sdk

2a35a5e

xxhZs and others added 2 commits December 1, 2023 13:32

fix timestamptz

0aa6785

use mod path to avoid macro

9b7b543

github-actions bot reviewed Dec 1, 2023

View reviewed changes

src/common/src/array/arrow/arrow_common.rs Outdated Show resolved Hide resolved

src/common/src/array/arrow/arrow_deltalake.rs Show resolved Hide resolved

src/common/src/array/arrow/mod.rs Outdated Show resolved Hide resolved

wenym1 reviewed Dec 7, 2023

View reviewed changes

e2e_test/sink/deltalake_rust_sink.slt Outdated Show resolved Hide resolved

xxhZs force-pushed the 13152-reimplement-delta-lake-sink-with-rust-sdk branch 4 times, most recently from de70454 to 8c41c0d Compare December 7, 2023 10:02

fix ci

5a492a6

fix ci fix ci fix fix ci fix ci fix

xxhZs force-pushed the 13152-reimplement-delta-lake-sink-with-rust-sdk branch from 8c41c0d to 5a492a6 Compare December 7, 2023 11:36

wenym1 reviewed Dec 8, 2023

View reviewed changes

fix

435eb73

wenym1 approved these changes Dec 12, 2023

View reviewed changes

e2e_test/sink/deltalake_rust_sink.slt Outdated Show resolved Hide resolved

xxhZs added 2 commits December 12, 2023 12:16

ut timeout 20 -> 22

cf96d64

fix region

4a83b8f

hzxa21 reviewed Dec 12, 2023

View reviewed changes

wenym1 added 2 commits December 13, 2023 02:37

reduce compile time and binary size

814dcc9

Merge branch 'main' into 13152-reimplement-delta-lake-sink-with-rust-sdk

f9605ad

temp add hakari third party

899070a

xxchan mentioned this pull request Dec 13, 2023

build: revisit the usage of hakari #13964

Closed

2 tasks

hzxa21 approved these changes Dec 13, 2023

View reviewed changes

use new delta-rs commit

86625ab

xxhZs added this pull request to the merge queue Dec 14, 2023

github-merge-queue bot removed this pull request from the merge queue due to failed status checks Dec 14, 2023

Merge branch 'main' into 13152-reimplement-delta-lake-sink-with-rust-sdk

5c365a9

xxhZs enabled auto-merge December 14, 2023 03:01

fix ci

a4023e5

xxhZs added this pull request to the merge queue Dec 14, 2023

Merged via the queue into main with commit cd04aaa Dec 14, 2023
29 of 30 checks passed

xxhZs deleted the 13152-reimplement-delta-lake-sink-with-rust-sdk branch December 14, 2023 03:44

xiangjinwu mentioned this pull request Dec 15, 2023

RUSTSEC-2018-0017: tempdir crate has been deprecated; use tempfile instead #14002

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(sink): support deltalake sink with rust sdk #13600

feat(sink): support deltalake sink with rust sdk #13600

xxhZs commented Nov 22, 2023 •

edited by wenym1

Loading

xxchan Nov 23, 2023 •

edited

Loading

xxhZs Nov 23, 2023

wangrunji0408 Nov 24, 2023

xxhZs Nov 24, 2023

wenym1 left a comment

wenym1 Nov 27, 2023

xxhZs Nov 28, 2023

wenym1 Dec 1, 2023

wenym1 left a comment

wenym1 Nov 30, 2023

xxhZs Nov 30, 2023 •

edited

Loading

wenym1 Dec 1, 2023

wenym1 commented Dec 1, 2023

xxhZs commented Dec 1, 2023

wenym1 commented Dec 1, 2023 •

edited

Loading

github-actions bot left a comment

wenym1 commented Dec 1, 2023

wenym1 left a comment

hzxa21 left a comment

wenym1 commented Dec 13, 2023

feat(sink): support deltalake sink with rust sdk #13600

feat(sink): support deltalake sink with rust sdk #13600

Conversation

xxhZs commented Nov 22, 2023 • edited by wenym1 Loading

What's changed and what's your intention?

Checklist

Documentation

Release note

xxchan Nov 23, 2023 • edited Loading

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

wenym1 left a 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

wenym1 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xxhZs Nov 30, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wenym1 commented Dec 1, 2023

xxhZs commented Dec 1, 2023

wenym1 commented Dec 1, 2023 • edited Loading

github-actions bot left a comment

Choose a reason for hiding this comment

wenym1 commented Dec 1, 2023

wenym1 left a comment

Choose a reason for hiding this comment

hzxa21 left a comment

Choose a reason for hiding this comment

wenym1 commented Dec 13, 2023

xxhZs commented Nov 22, 2023 •

edited by wenym1

Loading

xxchan Nov 23, 2023 •

edited

Loading

xxhZs Nov 30, 2023 •

edited

Loading

wenym1 commented Dec 1, 2023 •

edited

Loading