Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat(iceberg): support iceberg sink create table #18362

Merged
merged 18 commits into from
Sep 3, 2024
Merged

feat(iceberg): support iceberg sink create table #18362

merged 18 commits into from
Sep 3, 2024

Conversation

chenzl25
Copy link
Contributor

@chenzl25 chenzl25 commented Sep 2, 2024 •
edited
Loading

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

What's changed and what's your intention?

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.

  • Support create_table_if_not_exists for iceberg sink. If this parameter is specified as true, it will create an iceberg table automatically.

Example:

create sink sk from t_auto2
with (
    type='upsert',
    primary_key='a',
    connector = 'iceberg',
    catalog.type = 'glue',
    catalog.name = 'demo',
    warehouse.path = 's3://my-iceberg-bucket-5950/demo',
    s3.access.key = 'xxxxxxxxxx',
    s3.secret.key = 'xxxxxxxxxx',
    s3.region = 'ap-southeast-2',
    database.name='demo_db',
    table.name='t_auto2',
    create_table_if_not_exists='true',
);

@chenzl25 chenzl25 requested review from xxhZs, fuyufjh and wenym1 September 2, 2024 16:01
@graphite-app graphite-app bot requested a review from a team September 2, 2024 17:06
@chenzl25 chenzl25 requested a review from Li0k September 3, 2024 05:17
@chenzl25 chenzl25 added the user-facing-changes Contains changes that are visible to users label Sep 3, 2024
Li0k
Li0k approved these changes Sep 3, 2024
View reviewed changes
Copy link
Contributor

@Li0k Li0k left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Rest, LGTM

src/connector/src/sink/iceberg/mod.rs
.map_err(|e| SinkError::Iceberg(anyhow!(e)))
.context("failed to convert arrow schema to iceberg schema")?;

let location = {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nips: Can use a separate function for handle the path

src/connector/src/sink/iceberg/storage_catalog.rs
) -> iceberg::Result<Table> {
todo!()
let table_ident = TableIdent::new(namespace.clone(), creation.name.clone());
let table_path = {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ditto

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you can ignore this comments, I'm going to propose another PR to refactor it.

src/frontend/src/handler/create_sink.rs
@@ -342,6 +342,10 @@ pub async fn get_partition_compute_info(
async fn get_partition_compute_info_for_iceberg(
iceberg_config: &IcebergConfig,
) -> Result<Option<PartitionComputeInfo>> {
// TODO: check table if exists
if iceberg_config.create_table_if_not_exists {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why can do this?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because this is an optimization. We can always skip it. https://github.com/risingwavelabs/rfcs/blob/532d50c7f19ef940582390a17412e633e069449e/rfcs/0077-iceberg-sink-shuffle.md

@chenzl25 chenzl25 enabled auto-merge September 3, 2024 06:01
@chenzl25 chenzl25 added this pull request to the merge queue Sep 3, 2024
Merged via the queue into main with commit 5ab2a59 Sep 3, 2024
32 of 33 checks passed
@chenzl25 chenzl25 deleted the dylan/support_iceberg_sink_create_table branch September 3, 2024 07:02
@BugenZhao BugenZhao mentioned this pull request Dec 11, 2024
Closed
1 task
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@Li0k Li0k Li0k approved these changes

@xxhZs xxhZs Awaiting requested review from xxhZs

@fuyufjh fuyufjh Awaiting requested review from fuyufjh

@wenym1 wenym1 Awaiting requested review from wenym1

Assignees
No one assigned
Labels
ci/run-e2e-iceberg-sink-tests type/feature user-facing-changes Contains changes that are visible to users
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
2 participants
@chenzl25 @Li0k