Skip to content

Support insert into to seed kafka messages via risingwave binding #1275

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

Merged
jfallows merged 12 commits into from
Oct 3, 2024
Merged

Support insert into to seed kafka messages via risingwave binding #1275

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
cee6ba9
WIP
akrambek Sep 27, 2024
03ac3ee
Merge branch 'develop' into bug/risingwave-insert
akrambek Oct 1, 2024
1718164
WIP
akrambek Oct 1, 2024
38b7050
WIP
akrambek Oct 2, 2024
acbb805
WIP
akrambek Oct 2, 2024
69482d7
Rename table without primary to stream
akrambek Oct 2, 2024
fb055ed
Refactor test name
akrambek Oct 2, 2024
e40e291
Fix typo
akrambek Oct 2, 2024
b31bfeb
Checkpoint CREATE STREAM
akrambek Oct 2, 2024
c50840b
WIP
akrambek Oct 2, 2024
42ea5ad
Checkpoint
akrambek Oct 3, 2024
1888523
Fix regex
akrambek Oct 3, 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
Prev Previous commit
Next Next commit
WIP
  • Loading branch information
@akrambek
akrambek committed Oct 2, 2024
commit 38b7050cd1d0d6fbe2eb167e311e4316ffcbffb6
141 changes: 122 additions & 19 deletions ...inding/risingwave/streams/effective/create.table.with.primary.key.and.includes/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,25 +32,73 @@ write zilla:data.ext ${pgsql:dataEx()
.query()
.build()
.build()}
write "CREATE TABLE IF NOT EXISTS cities (\n"
" *,\n"
" PRIMARY KEY (id)\n"
")\n"
"INCLUDE KEY AS key\n"
"INCLUDE header 'zilla:correlation-id' AS correlation_id\n"
"INCLUDE timestamp AS timestamp\n"
"INCLUDE header 'zilla:identity' AS owner_id\n"
"WITH (\n"
" connector='kafka',\n"
" properties.bootstrap.server='localhost:9092',\n"
" topic='dev.cities',\n"
" scan.startup.mode='latest',\n"
" scan.startup.timestamp.millis='140000000'\n"
") FORMAT PLAIN ENCODE AVRO (\n"
" schema.registry = 'http://localhost:8081'\n"
");"
write "CREATE SOURCE IF NOT EXISTS cities_source (*)\n"
"INCLUDE header 'zilla:correlation-id' AS zilla_correlation_id_header,\n"
"INCLUDE timestamp AS timestamp,\n"
"INCLUDE header 'zilla:identity' AS zilla_identity_header,\n"
"WITH (\n"
" connector='kafka',\n"
" properties.bootstrap.server='localhost:9092',\n"
" topic='dev.cities',\n"
" scan.startup.mode='latest',\n"
" scan.startup.timestamp.millis='140000000'\n"
") FORMAT PLAIN ENCODE AVRO (\n"
" schema.registry = 'http://localhost:8081'\n"
");"
[0x00]
write flush

read advised zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
.completion()
.tag("CREATE_SOURCE")
.build()
.build()}

read advised zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
.ready()
.status("IDLE")
.build()
.build()}

write zilla:data.ext ${pgsql:dataEx()
.typeId(zilla:id("pgsql"))
.query()
.build()
.build()}
write "CREATE MATERIALIZED VIEW cities_view AS\n"
" SELECT id, name, description,"
" COALESCE(correlation_id, zilla_correlation_id_header::varchar) as correlation_id,"
" COALESCE(identity, zilla_identity_header::varchar) as identity,
" timestamp FROM cities_source;"
[0x00]
write flush

read advised zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
.completion()
.tag("CREATE_MATERIALIZED_VIEW")
.build()
.build()}

read advised zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
.ready()
.status("IDLE")
.build()
.build()}

write zilla:data.ext ${pgsql:dataEx()
.typeId(zilla:id("pgsql"))
.query()
.build()
.build()}
write "CREATE TABLE IF NOT EXISTS cities "
"(id VARCHAR, name VARCHAR, description VARCHAR,"
" correlation_id VARCHAR, identity as VARCHAR, timestamp TIMESTAMP,"
" PRIMARY KEY (id));"
[0x00]
write flush

read advised zilla:flush ${pgsql:flushEx()
Expand All @@ -67,12 +115,65 @@ read advised zilla:flush ${pgsql:flushEx()
.build()
.build()}

write zilla:data.ext ${pgsql:dataEx()
.typeId(zilla:id("pgsql"))
.query()
.build()
.build()}
write "CREATE SINK cities_view_sink INTO cities FROM cities_view;"
[0x00]
write flush

read advised zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
.completion()
.tag("CREATE_SINK")
.build()
.build()}

read advised zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
.ready()
.status("IDLE")
.build()
.build()}

write zilla:data.ext ${pgsql:dataEx()
.typeId(zilla:id("pgsql"))
.query()
.build()
.build()}
write "CREATE SINK cities_sink\n"
"FROM cities\n"
"WITH (\n"
" connector='kafka',\n"
" topic='dev.cities',\n"
" properties.bootstrap.server='localhost:9092',\n"
" primary_key='id'\n
") FORMAT UPSERT ENCODE AVRO (\n"
" schema.registry = 'http://localhost:8081'\n"
");"
[0x00]
write flush

read advised zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
.completion()
.tag("CREATE_SINK")
.build()
.build()}

read advised zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
.ready()
.status("IDLE")
.build()
.build()}

connect "zilla://streams/app2"
option zilla:window 8192
option zilla:transmission "duplex"


write zilla:begin.ext ${pgsql:beginEx()
.typeId(zilla:id("pgsql"))
.parameter("user", "root")
Expand All @@ -89,7 +190,9 @@ write zilla:data.ext ${pgsql:dataEx()
.build()
.build()}
write "CREATE TOPIC IF NOT EXISTS cities "
"(id VARCHAR, name VARCHAR, description VARCHAR, PRIMARY KEY (id));"
"(id VARCHAR, name VARCHAR, description VARCHAR,"
" correlation_id VARCHAR, identity as VARCHAR, timestamp TIMESTAMP,"
" PRIMARY KEY (id));"
[0x00]
write flush

Expand Down
5 changes: 1 addition & 4 deletions ...inding/risingwave/streams/effective/create.table.with.primary.key.and.includes/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,8 +70,6 @@ read zilla:data.ext ${pgsql:dataEx()
.query()
.build()
.build()}
read "CREATE MATERIALIZED VIEW cities_view AS SELECT * FROM cities_source;"
[0x00]
read "CREATE MATERIALIZED VIEW cities_view AS\n"
" SELECT id, name, description,"
" COALESCE(correlation_id, zilla_correlation_id_header::varchar) as correlation_id,"
Expand Down Expand Up @@ -171,7 +169,6 @@ write advise zilla:flush ${pgsql:flushEx()
.build()
.build()}


accept "zilla://streams/app2"
option zilla:window 8192
option zilla:transmission "duplex"
Expand All @@ -195,7 +192,7 @@ read zilla:data.ext ${pgsql:dataEx()
.build()}
read "CREATE TOPIC IF NOT EXISTS cities "
"(id VARCHAR, name VARCHAR, description VARCHAR,"
" correlation_id VARCHAR, identity as VARCHAR, timestamp TIMESTAMP"
" correlation_id VARCHAR, identity as VARCHAR, timestamp TIMESTAMP,"
" PRIMARY KEY (id));"
[0x00]

Expand Down
1 change: 0 additions & 1 deletion ...ipts/io/aklivity/zilla/specs/binding/risingwave/streams/effective/create.table/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,6 @@ write advise zilla:flush ${pgsql:flushEx()
.build()
.build()}


accept "zilla://streams/app2"
option zilla:window 8192
option zilla:transmission "duplex"
Expand Down
Loading