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
Next Next commit
WIP
  • Loading branch information
@akrambek
akrambek committed Sep 27, 2024
commit cee6ba965eb8656dde7d73b71cdf28d23fb09d4c
121 changes: 115 additions & 6 deletions ...ty/zilla/specs/binding/risingwave/streams/effective/create.table.with.includes/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,10 @@ write zilla:data.ext ${pgsql:dataEx()
.query()
.build()
.build()}
write "CREATE SOURCE IF NOT EXISTS weather (*)\n"
"INCLUDE header 'zilla:correlation-id' AS correlation_id\n"
"INCLUDE timestamp AS timestamp\n"
"INCLUDE header 'zilla:identity' AS owner_id\n"
write "CREATE SOURCE IF NOT EXISTS weather_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"
Expand All @@ -62,6 +62,114 @@ read advised zilla:flush ${pgsql:flushEx()
.build()
.build()}

write zilla:data.ext ${pgsql:dataEx()
.typeId(zilla:id("pgsql"))
.query()
.build()
.build()}
write "CREATE MATERIALIZED VIEW weather_view AS\n"
" SELECT city, temperature, date,"
" COALESCE(correlation_id, zilla_correlation_id_header::varchar) as correlation_id,"
" COALESCE(identity, zilla_identity_header::varchar) as identity,
" timestamp FROM streampay_users_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 weather("
" city VARCHAR, temperature DOUBLE, date DATE,
" correlation_id VARCHAR, identity as VARCHAR, timestamp TIMESTAMP,"
" PRIMARY KEY (city, temperature, date));"
[0x00]
write flush

read advised zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
.completion()
.tag("CREATE_TABLE")
.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 weather_view_sink INTO weather FROM weather_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 weather_sink\n"
"FROM weather\n"
"WITH (\n"
" connector='kafka',\n"
" topic='dev.weather',\n"
" properties.bootstrap.server='localhost:9092',\n"
") FORMAT PLAIN ENCODE AVRO (\n"
" schema.registry = 'http://localhost:8081',\n"
" force_append_only='true'\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"
Expand All @@ -81,8 +189,9 @@ write zilla:data.ext ${pgsql:dataEx()
.query()
.build()
.build()}
write "CREATE TOPIC IF NOT EXISTS weather "
"(city VARCHAR, temperature DOUBLE, date DATE);"
write "CREATE TOPIC IF NOT EXISTS weather"
" (city VARCHAR, temperature DOUBLE, date DATE,"
" correlation_id VARCHAR, identity as VARCHAR, timestamp TIMESTAMP);"
[0x00]

write flush
Expand Down
137 changes: 121 additions & 16 deletions ...ty/zilla/specs/binding/risingwave/streams/effective/create.table.with.includes/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,20 +36,20 @@ read zilla:data.ext ${pgsql:dataEx()
.query()
.build()
.build()}
read "CREATE SOURCE IF NOT EXISTS weather (*)\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.weather',\n"
" scan.startup.mode='latest',\n"
" scan.startup.timestamp.millis='140000000'\n"
") FORMAT PLAIN ENCODE AVRO (\n"
" schema.registry = 'http://localhost:8081'\n"
");"
[0x00]
read "CREATE SOURCE IF NOT EXISTS weather_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.weather',\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 advise zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
Expand All @@ -65,6 +65,110 @@ write advise zilla:flush ${pgsql:flushEx()
.build()
.build()}

read zilla:data.ext ${pgsql:dataEx()
.typeId(zilla:id("pgsql"))
.query()
.build()
.build()}
read "CREATE MATERIALIZED VIEW weather_view AS\n"
" SELECT city, temperature, date,"
" COALESCE(correlation_id, zilla_correlation_id_header::varchar) as correlation_id,"
" COALESCE(identity, zilla_identity_header::varchar) as identity,
" timestamp FROM streampay_users_source;"
[0x00]

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

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

read zilla:data.ext ${pgsql:dataEx()
.typeId(zilla:id("pgsql"))
.query()
.build()
.build()}
read "CREATE TABLE weather("
" city VARCHAR, temperature DOUBLE, date DATE,
" correlation_id VARCHAR, identity as VARCHAR, timestamp TIMESTAMP,"
" PRIMARY KEY (city, temperature, date));"
[0x00]

write advise zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
.completion()
.tag("CREATE_TABLE")
.build()
.build()}

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

read zilla:data.ext ${pgsql:dataEx()
.typeId(zilla:id("pgsql"))
.query()
.build()
.build()}
read "CREATE SINK weather_view_sink INTO weather FROM weather_view;"
[0x00]

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

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

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

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

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


accept "zilla://streams/app2"
option zilla:window 8192
Expand All @@ -87,8 +191,9 @@ read zilla:data.ext ${pgsql:dataEx()
.query()
.build()
.build()}
read "CREATE TOPIC IF NOT EXISTS weather "
"(city VARCHAR, temperature DOUBLE, date DATE);"
read "CREATE TOPIC IF NOT EXISTS weather"
" (city VARCHAR, temperature DOUBLE, date DATE,"
" correlation_id VARCHAR, identity as VARCHAR, timestamp TIMESTAMP);"
[0x00]

write advise zilla:flush ${pgsql:flushEx()
Expand Down
Loading