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

feat(sink): Support Sinking Enum16 Ids to Clickhouse #14668

Merged
merged 10 commits into from
Feb 26, 2024
Merged

feat(sink): Support Sinking Enum16 Ids to Clickhouse #14668

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
6e0149e
Support Inserting Enum16 Ids
HurricanKai Jan 18, 2024
5705931
Clickhouse Enum E2E
HurricanKai Jan 20, 2024
8791c65
Fix Trailing Space
HurricanKai Jan 20, 2024
a443b60
Merge branch 'main' into main
xxhZs Jan 22, 2024
7927d89
Fix Typo
HurricanKai Jan 23, 2024
e60d9e6
Merge remote-tracking branch 'refs/remotes/origin/main'
HurricanKai Jan 23, 2024
a5c2b07
Fix Enum Check ordering
HurricanKai Feb 9, 2024
bcb7492
Merge branch 'main' into main
HurricanKai Feb 10, 2024
c5fd0af
Merge branch 'main' into main
HurricanKai Feb 19, 2024
a80ff77
Merge branch 'main' into main
stdrc Feb 21, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions ci/scripts/e2e-clickhouse-sink-test.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ sleep 1
echo "--- create clickhouse table"
curl https://clickhouse.com/ | sh
sleep 2
./clickhouse client --host=clickhouse-server --port=9000 --query="CREATE table demo_test(v1 Int32,v2 Int64,v3 String)ENGINE = ReplacingMergeTree PRIMARY KEY (v1);"
./clickhouse client --host=clickhouse-server --port=9000 --query="CREATE table demo_test(v1 Int32,v2 Int64,v3 String,v4 Enum16('A'=1,'B'=2))ENGINE = ReplacingMergeTree PRIMARY KEY (v1);"

echo "--- testing sinks"
sqllogictest -p 4566 -d dev './e2e_test/sink/clickhouse_sink.slt'
Expand All @@ -41,13 +41,13 @@ sleep 1

# check sink destination using shell
if cat ./query_result.csv | sort | awk -F "," '{
if ($1 == 1 && $2 == 50 && $3 == "\"1-50\"") c1++;
if ($1 == 13 && $2 == 2 && $3 == "\"13-2\"") c2++;
if ($1 == 2 && $2 == 2 && $3 == "\"2-2\"") c3++;
if ($1 == 21 && $2 == 2 && $3 == "\"21-2\"") c4++;
if ($1 == 3 && $2 == 2 && $3 == "\"3-2\"") c5++;
if ($1 == 5 && $2 == 2 && $3 == "\"5-2\"") c6++;
if ($1 == 8 && $2 == 2 && $3 == "\"8-2\"") c7++; }
if ($1 == 1 && $2 == 50 && $3 == "\"1-50\"" && $4 == "\"A\"") c1++;
if ($1 == 13 && $2 == 2 && $3 == "\"13-2\"" && $4 == "\"B\"") c2++;
if ($1 == 2 && $2 == 2 && $3 == "\"2-2\"" && $4 == "\"A\"") c3++;
HurricanKai marked this conversation as resolved.
Show resolved Hide resolved
if ($1 == 21 && $2 == 2 && $3 == "\"21-2\"" && $4 == "\"B\"") c4++;
HurricanKai marked this conversation as resolved.
Show resolved Hide resolved
if ($1 == 3 && $2 == 2 && $3 == "\"3-2\"" && $4 == "\"A\"") c5++;
if ($1 == 5 && $2 == 2 && $3 == "\"5-2\"" && $4 == "\"B\"") c6++;
if ($1 == 8 && $2 == 2 && $3 == "\"8-2\"" && $4 == "\"A\"") c7++; }
END { exit !(c1 == 1 && c2 == 1 && c3 == 1 && c4 == 1 && c5 == 1 && c6 == 1 && c7 == 1); }'; then
echo "Clickhouse sink check passed"
else
Expand Down
6 changes: 3 additions & 3 deletions e2e_test/sink/clickhouse_sink.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,11 @@
statement ok
CREATE TABLE t6 (v1 int primary key, v2 bigint, v3 varchar);
CREATE TABLE t6 (v1 int primary key, v2 bigint, v3 varchar, v4 smallint);

statement ok
CREATE MATERIALIZED VIEW mv6 AS SELECT * FROM t6;

statement ok
CREATE SINK s6 AS select mv6.v1 as v1, mv6.v2 as v2, mv6.v3 as v3 from mv6 WITH (
CREATE SINK s6 AS select mv6.v1 as v1, mv6.v2 as v2, mv6.v3 as v3, mv6.v4 as v4 from mv6 WITH (
connector = 'clickhouse',
type = 'append-only',
force_append_only='true',
Expand All @@ -17,7 +17,7 @@ CREATE SINK s6 AS select mv6.v1 as v1, mv6.v2 as v2, mv6.v3 as v3 from mv6 WITH
);

statement ok
INSERT INTO t6 VALUES (1, 50, '1-50'), (2, 2, '2-2'), (3, 2, '3-2'), (5, 2, '5-2'), (8, 2, '8-2'), (13, 2, '13-2'), (21, 2, '21-2');
INSERT INTO t6 VALUES (1, 50, '1-50', 1), (2, 2, '2-2', 2), (3, 2, '3-2', 1), (5, 2, '5-2', 2), (8, 2, '8-2', 1), (13, 2, '13-2', 2), (21, 2, '21-2', 1);

statement ok
FLUSH;
Expand Down
8 changes: 5 additions & 3 deletions src/connector/src/sink/clickhouse.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,9 +260,11 @@ impl ClickHouseSink {
) -> Result<()> {
let is_match = match fields_type {
risingwave_common::types::DataType::Boolean => Ok(ck_column.r#type.contains("Bool")),
risingwave_common::types::DataType::Int16 => {
Ok(ck_column.r#type.contains("UInt16") | ck_column.r#type.contains("Int16"))
}
risingwave_common::types::DataType::Int16 => Ok(ck_column.r#type.contains("UInt16")
HurricanKai marked this conversation as resolved.
Show resolved Hide resolved
| ck_column.r#type.contains("Int16")
// Allow Int16 to be pushed to Enum16, they share an encoding and value range
// No special care is taken to ensure values are valid.
| ck_column.r#type.contains("Enum16")),
risingwave_common::types::DataType::Int32 => {
Ok(ck_column.r#type.contains("UInt32") | ck_column.r#type.contains("Int32"))
}
Expand Down
Loading