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

Propagate error code in risingwave binding that's coming either from pgsql-kafka or risingwave #1288

Merged
merged 3 commits into from
Oct 9, 2024
Merged

Propagate error code in risingwave binding that's coming either from pgsql-kafka or risingwave #1288

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ec232cd
WIP
akrambek Oct 9, 2024
1a57d0b
Checkpoint
akrambek Oct 9, 2024
652a5c4
WIP
akrambek Oct 9, 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
46 changes: 46 additions & 0 deletions ...inding-pgsql.spec/src/main/java/io/aklivity/zilla/specs/binding/pgsql/PgsqlFunctions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlColumnInfoFW;
import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlCompletedFlushExFW;
import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlDataExFW;
import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlErrorFlushExFW;
import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlFlushExFW;
import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlFormat;
import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlMessageType;
Expand Down Expand Up @@ -215,6 +216,13 @@ public PgsqlCompletedFlushExBuilder completion()
return new PgsqlCompletedFlushExBuilder();
}

public PgsqlErrorFlushExBuilder error()
{
flushExRW.kind(PgsqlMessageType.ERROR.value());

return new PgsqlErrorFlushExBuilder();
}

public PgsqlReadyFlushExBuilder ready()
{
flushExRW.kind(PgsqlMessageType.READY.value());
Expand Down Expand Up @@ -351,6 +359,44 @@ public PgsqlFlushExBuilder build()
}
}

public final class PgsqlErrorFlushExBuilder
{
private final PgsqlErrorFlushExFW.Builder pgsqlErrorFlushExRW = new PgsqlErrorFlushExFW.Builder();

private PgsqlErrorFlushExBuilder()
{
pgsqlErrorFlushExRW.wrap(writeBuffer, PgsqlFlushExFW.FIELD_OFFSET_TYPE, writeBuffer.capacity());
}

public PgsqlErrorFlushExBuilder severity(
String severity)
{
pgsqlErrorFlushExRW.severity(String.format("%s\u0000", severity));
return this;
}

public PgsqlErrorFlushExBuilder code(
String code)
{
pgsqlErrorFlushExRW.code(String.format("%s\u0000", code));
return this;
}

public PgsqlErrorFlushExBuilder message(
String message)
{
pgsqlErrorFlushExRW.message(String.format("%s\u0000", message));
return this;
}

public PgsqlFlushExBuilder build()
{
final PgsqlErrorFlushExFW pgsqlErrorFlushEx = pgsqlErrorFlushExRW.build();
flushExRO.wrap(writeBuffer, 0, pgsqlErrorFlushEx.limit());
return PgsqlFlushExBuilder.this;
}
}

public final class PgsqlReadyFlushExBuilder
{
private final PgsqlReadyFlushExFW.Builder pgsqlReadyFlushExRW = new PgsqlReadyFlushExFW.Builder();
Expand Down
9 changes: 9 additions & 0 deletions incubator/binding-pgsql.spec/src/main/resources/META-INF/zilla/pgsql.idl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ scope pgsql
ROW (68),
TYPE (84),
COMPLETION (67),
ERROR (69),
READY (90)
}

Expand All @@ -56,6 +57,7 @@ scope pgsql
{
case 84: pgsql::stream::PgsqlTypeFlushEx type;
case 67: pgsql::stream::PgsqlCompletedFlushEx completion;
case 69: pgsql::stream::PgsqlErrorFlushEx error;
case 90: pgsql::stream::PgsqlReadyFlushEx ready;
}

Expand Down Expand Up @@ -86,6 +88,13 @@ scope pgsql
string16 tag;
}

struct PgsqlErrorFlushEx
{
string16 severity;
string16 code;
string16 message;
}

enum PgsqlStatus (uint8)
{
IDLE(73),
Expand Down
56 changes: 56 additions & 0 deletions ...aklivity/zilla/specs/binding/pgsql/streams/application/select.table.with.error/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
#
# Copyright 2021-2023 Aklivity Inc
#
# Licensed under the Aklivity Community License (the "License"); you may not use
# this file except in compliance with the License. You may obtain a copy of the
# License at
#
# https://www.aklivity.io/aklivity-community-license/
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
#

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

write zilla:begin.ext ${pgsql:beginEx()
.typeId(zilla:id("pgsql"))
.parameter("user", "root")
.parameter("database", "dev")
.parameter("application_name", "psql")
.parameter("client_encoding", "UTF8")
.build()}

connected

write zilla:data.ext ${pgsql:dataEx()
.typeId(zilla:id("pgsql"))
.query()
.build()
.build()}
write "SELECT * FROM unknown;"
[0x00]
write flush

read advised zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
.error()
.severity("ERROR")
.code("XX000")
.message("Failed to run the query.")
.build()
.build()}

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

read closed
write close
59 changes: 59 additions & 0 deletions ...aklivity/zilla/specs/binding/pgsql/streams/application/select.table.with.error/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
#
# Copyright 2021-2023 Aklivity Inc
#
# Licensed under the Aklivity Community License (the "License"); you may not use
# this file except in compliance with the License. You may obtain a copy of the
# License at
#
# https://www.aklivity.io/aklivity-community-license/
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
#

property serverAddress "zilla://streams/app0"

accept ${serverAddress}
option zilla:window 8192
option zilla:transmission "duplex"

accepted

read zilla:begin.ext ${pgsql:beginEx()
.typeId(zilla:id("pgsql"))
.parameter("user", "root")
.parameter("database", "dev")
.parameter("application_name", "psql")
.parameter("client_encoding", "UTF8")
.build()}

connected

read zilla:data.ext ${pgsql:dataEx()
.typeId(zilla:id("pgsql"))
.query()
.build()
.build()}
read "SELECT * FROM unknown;"
[0x00]

write advise zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
.error()
.severity("ERROR")
.code("XX000")
.message("Failed to run the query.")
.build()
.build()}

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

write close
read closed
93 changes: 93 additions & 0 deletions .../io/aklivity/zilla/specs/binding/pgsql/streams/network/select.table.with.error/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
#
# Copyright 2021-2023 Aklivity Inc
#
# Licensed under the Aklivity Community License (the "License"); you may not use
# this file except in compliance with the License. You may obtain a copy of the
# License at
#
# https://www.aklivity.io/aklivity-community-license/
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
#

property networkConnectWindow 8192

connect "zilla://streams/net0"
option zilla:window ${networkConnectWindow}
option zilla:transmission "duplex"
option zilla:byteorder "network"

connected

write 75 # length
3s # major version
0s # minor version
"user" [0x00] # name
"root" [0x00] # value
"database" [0x00] # name
"dev" [0x00] # value
"application_name" [0x00] # name
"psql" [0x00] # value
"client_encoding" [0x00] # name
"UTF8" [0x00] # value
[0x00] # end of parameters

read [0x52] # type R
8 # length
0 # authentication type

read [0x4b] # type K
12 # length
0 # pid
0 # key

read [0x53] # type S
25 # length
"client_encoding" [0x00] # name
"UTF8" [0x00] # value

read [0x53] # type S
35 # length
"standard_conforming_strings" [0x00] # name
[0x6f 0x6e 0x00] # value

read [0x53] # type S
25 # length
"server_version" [0x00] # name
"1.0.0" [0x00] # value

read [0x53] # type S
27 # length
"application_name" [0x00] # name
"zilla" [0x00] # value

read [0x5a] # type Z
5 # length
[0x49] # status

write [0x51] # type Q
27 # length
"SELECT * FROM unknown;"
[0x00]

read [0x45] # type E
41 # length
"ERROR" # severity
[0x00] # end of string
"XX000" # code
[0x00] # end of string
"Failed to run the query." # code
[0x00] # end of string

read [0x5a] # type Z
5 # length
[0x49] # status

read [0x58] # type X
4 # length

read closed
write close
Loading