feat: support precompute partition for iceberg #14710

ZENOTME · 2024-01-22T04:48:38Z

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

What's changed and what's your intention?

This is the final part of #13898:

support get partition info at frontend
add the project node to compute the partition value and exchange node to shuffle by it

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.

liurenjie1024

Generally LGTM, thanks!

proto/stream_plan.proto

src/connector/src/sink/catalog/desc.rs

liurenjie1024 · 2024-01-23T02:21:29Z

src/connector/src/sink/iceberg/mod.rs

@@ -177,7 +178,7 @@ impl IcebergConfig {
        self.catalog_type.as_deref().unwrap_or("storage")
    }

-    fn build_iceberg_configs(&self) -> Result<HashMap<String, String>> {
+    pub fn build_iceberg_configs(&self) -> Result<HashMap<String, String>> {


Why we need to make this pub?

liurenjie1024 · 2024-01-23T02:21:41Z

src/connector/src/sink/iceberg/mod.rs

@@ -346,6 +347,55 @@ impl IcebergConfig {
    }
 }

+pub async fn create_catalog(config: &IcebergConfig) -> Result<CatalogRef> {


Suggested change

pub async fn create_catalog(config: &IcebergConfig) -> Result<CatalogRef> {

async fn create_catalog(config: &IcebergConfig) -> Result<CatalogRef> {

src/connector/src/sink/iceberg/mod.rs

liurenjie1024 · 2024-01-23T02:22:13Z

src/connector/src/sink/iceberg/mod.rs

-            .load_table(&table_id)
-            .await
-            .map_err(|err| SinkError::Iceberg(anyhow!(err)))?;
+    async fn create_validated_table(&self) -> Result<Table> {


Suggested change

async fn create_validated_table(&self) -> Result<Table> {

async fn create_and_validate_table(&self) -> Result<Table> {

liurenjie1024 · 2024-01-23T02:47:47Z

src/frontend/src/handler/create_sink.rs

+    let table = create_table(iceberg_config).await?;
+    let partition_spec = table
+        .current_table_metadata()
+        .current_partition_spec()


We should not throw error here, user may create a table without partition.

src/frontend/src/optimizer/plan_node/stream_sink.rs

liurenjie1024 · 2024-01-23T08:01:37Z

src/frontend/src/optimizer/plan_node/stream_sink.rs

@@ -133,9 +236,42 @@ impl StreamSink {
        Ok(Self::new(input, sink))
    }

+    fn derive_iceberg_sink_distribution(


Could we add some planner tests for this? I think it would better to ensure its consistency.

I find it inconvenient to test it for Iceberg because I need to connect the catalog and get related info.
We test the sink using handle_explain, which means that there isn't a place to insert the mock partition info in plan test.

One way to test it is to mock the partition info in blackhole connector. But this will add the code for test in frontend, I'm not sure whether it's a good way.

pub async fn get_partition_compute_info( with_options: &WithOptions, ) -> Result<Option<PartitionComputeInfo>> { match connector.as_str() { ICEBERG_SINK => { .. } BLACK_HOLE => { ..//generate mock info } _ => Ok(None), } }

Another way is to modify the test way for sink, maybe we can use create_sink instead of test using explain and add a param to pass the partition info so that we can mock the partition info outside.

risingwave/src/frontend/planner_test/src/lib.rs

Line 550 in ee08ac5

explain::handle_explain(handler_args, *statement, options, analyze).await?;

I find it inconvenient to test it for Iceberg because I need to connect the catalog and get related info.

How about creating a mocked iceberg catalog?

BugenZhao · 2024-01-23T08:06:02Z

Is this PR specific to Iceberg? Would you please update the PR title to make it more detailed?

ZENOTME · 2024-01-23T08:07:25Z

Is this PR specific to Iceberg? Would you please update the PR title to make it more detailed?

Sorry. Update it.

github-actions

license-eye has totally checked 4768 files.

Valid	Invalid	Ignored	Fixed
2077	2	2689	0

Click to see the invalid file list

src/common/src/array/arrow/arrow_iceberg.rs
src/connector/src/sink/iceberg/mock_catalog.rs

src/common/src/array/arrow/arrow_iceberg.rs

src/connector/src/sink/iceberg/mock_catalog.rs

liurenjie1024 · 2024-01-26T03:57:40Z

src/frontend/planner_test/tests/testdata/output/sink.yaml

@@ -172,3 +172,45 @@
    StreamSink { type: upsert, columns: [a, b, t2._row_id(hidden)], pk: [t2._row_id] }
    └─StreamEowcSort { sort_column: t2.b }
      └─StreamTableScan { table: t2, columns: [a, b, _row_id] }
+- id: create_mock_iceberg_sink_append_only_with_partition


This test cases seem not enough, in the rfc there are 4 cases, please add test for each case.

tabVersion

not familiar with the context, waiting for renjie's approval

tabVersion · 2024-01-29T03:55:36Z

src/frontend/src/optimizer/plan_node/stream_sink.rs

+/// ## Why we need `PartitionComputeInfo`?
+///
+/// For some sink, it will write the data into different file based on the partition value. E.g. iceberg sink(<https://iceberg.apache.org/spec/#partitioning>)
+/// For this kind of sink, the file num can be reduced if we can shuffle the data based on the partition value. More details can be found in <https://github.com/risingwavelabs/rfcs/pull/77>.


does this also apply to other lakehouse downstream?

Yes, seems delta also has partition concept. I guess it can apply it.🤔

tabVersion · 2024-01-29T04:02:58Z

src/frontend/src/handler/create_sink.rs

+    // Separate the partition spec into two parts: sparse partition and range partition.
+    // Sparse partition means that the data distribution is more sparse at a given time.
+    // Range partition means that the data distribution is likely same at a given time.
+    // Only compute the partition and shuffle by them for the sparse partition.


so we are changing the distribution key to adjust each CN's hash range?

Yes, we can see that for sparse partition, we will add the exchange before sink.

risingwave/src/frontend/planner_test/tests/testdata/output/sink.yaml

Line 233 in 8b263cb

└─StreamExchange { dist: HashShard($expr1) }

liurenjie1024

LGTM, thanks!

liurenjie1024 · 2024-01-31T03:37:02Z

cc @xxchan @BugenZhao PTAL for cargo.lock

hzxa21

LGTM for Cargo.lock

ZENOTME requested a review from a team as a code owner January 22, 2024 04:48

github-actions bot added the type/feature label Jan 22, 2024

ZENOTME added ci/run-e2e-iceberg-sink-tests and removed type/feature labels Jan 22, 2024

ZENOTME force-pushed the zj/partition_compute_final branch 2 times, most recently from 83392c9 to bc926bd Compare January 22, 2024 08:13

ZENOTME requested review from liurenjie1024, tabVersion, shanicky and xxchan January 22, 2024 09:27

liurenjie1024 reviewed Jan 23, 2024

View reviewed changes

ZENOTME changed the title ~~feat: support precompute partition~~ feat: support precompute partition for iceberg Jan 23, 2024

ZENOTME added 5 commits January 24, 2024 22:31

support precompute partition

065898e

fix

44861f4

add ut test && plan test

3c4f716

fix and refine e2e test

60641e8

refine

6d18a72

ZENOTME force-pushed the zj/partition_compute_final branch from b46fa78 to 6d18a72 Compare January 24, 2024 14:31

github-actions bot reviewed Jan 24, 2024

View reviewed changes

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

src/connector/src/sink/iceberg/mock_catalog.rs Show resolved Hide resolved

fix license

c99f87e

liurenjie1024 reviewed Jan 26, 2024

View reviewed changes

refine plan test

0ee0497

tabVersion reviewed Jan 29, 2024

View reviewed changes

add e2e test

397faa4

ZENOTME force-pushed the zj/partition_compute_final branch from 8b263cb to 397faa4 Compare January 29, 2024 07:23

liurenjie1024 approved these changes Jan 30, 2024

View reviewed changes

liurenjie1024 requested a review from BugenZhao January 31, 2024 03:37

liurenjie1024 requested a review from hzxa21 January 31, 2024 11:35

hzxa21 approved these changes Jan 31, 2024

View reviewed changes

liurenjie1024 enabled auto-merge January 31, 2024 11:46

github-actions bot added type/feature labels Jan 31, 2024

auto-merge was automatically disabled January 31, 2024 23:49
Merge queue setting changed

ZENOTME added this pull request to the merge queue Feb 1, 2024

Merged via the queue into main with commit 7d0d43f Feb 1, 2024
42 of 43 checks passed

ZENOTME deleted the zj/partition_compute_final branch February 1, 2024 03:13

liurenjie1024 mentioned this pull request Feb 1, 2024

Add shuffle in iceberg sink to reduce small files as much as possible. #13898

Closed

cyliu0 mentioned this pull request Feb 2, 2024

v1.7.0-rc/nightly-20240201 source throughput down to 0 with non-shared PG CDC sources #14943

Closed

ZENOTME mentioned this pull request Apr 1, 2024

feat: draft stream partition compute #14135

Closed

9 tasks

chenzl25 mentioned this pull request Aug 22, 2024

bug: iceberg sink precompute partition key would generate wrong data if table partition key changed #18185

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: support precompute partition for iceberg #14710

feat: support precompute partition for iceberg #14710

ZENOTME commented Jan 22, 2024 •

edited by liurenjie1024

Loading

liurenjie1024 left a comment

liurenjie1024 Jan 23, 2024

liurenjie1024 Jan 23, 2024

liurenjie1024 Jan 23, 2024

liurenjie1024 Jan 23, 2024

liurenjie1024 Jan 23, 2024

ZENOTME Jan 24, 2024

liurenjie1024 Jan 24, 2024

BugenZhao commented Jan 23, 2024

ZENOTME commented Jan 23, 2024

github-actions bot left a comment

liurenjie1024 Jan 26, 2024

tabVersion left a comment

tabVersion Jan 29, 2024

ZENOTME Jan 29, 2024

tabVersion Jan 29, 2024

ZENOTME Jan 29, 2024

liurenjie1024 left a comment

liurenjie1024 commented Jan 31, 2024

hzxa21 left a comment

	pub async fn create_catalog(config: &IcebergConfig) -> Result<CatalogRef> {
	async fn create_catalog(config: &IcebergConfig) -> Result<CatalogRef> {

	async fn create_validated_table(&self) -> Result<Table> {
	async fn create_and_validate_table(&self) -> Result<Table> {

feat: support precompute partition for iceberg #14710

feat: support precompute partition for iceberg #14710

Conversation

ZENOTME commented Jan 22, 2024 • edited by liurenjie1024 Loading

What's changed and what's your intention?

Checklist

Documentation

Release note

liurenjie1024 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

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

BugenZhao commented Jan 23, 2024

ZENOTME commented Jan 23, 2024

github-actions bot left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tabVersion 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

Choose a reason for hiding this comment

liurenjie1024 left a comment

Choose a reason for hiding this comment

liurenjie1024 commented Jan 31, 2024

hzxa21 left a comment

Choose a reason for hiding this comment

ZENOTME commented Jan 22, 2024 •

edited by liurenjie1024

Loading