Support pgsql binding #1200

akrambek · 2024-08-14T21:30:00Z

Support psql binding with server and client kinds, so that pgsql client can be proxied to a pgsql server.

jfallows · 2024-08-14T22:39:27Z

...ts/io/aklivity/zilla/specs/binding/pgsql/streams/network/create.table/primary.key/client.rpt

+
+read 43                                 # type
+     17                                 # length
+     "CREATE_TABLE"                     # tag


We need the ready message after completed too.

jfallows · 2024-08-14T22:40:13Z

...ts/io/aklivity/zilla/specs/binding/pgsql/streams/network/create.table/primary.key/client.rpt

Let's rename the scenario to create.table.with.primary.key and avoid the nested directory structure.

jfallows · 2024-08-14T22:40:21Z

...o/aklivity/zilla/specs/binding/pgsql/streams/application/create.table/primary.key/client.rpt

Let's rename the scenario to create.table.with.primary.key and avoid the nested directory structure.

jfallows · 2024-08-14T22:40:49Z

...o/aklivity/zilla/specs/binding/pgsql/streams/application/create.table/primary.key/client.rpt

+
+write zilla:data.ext ${pgsql:dataEx()
+                              .typeId(zilla:id("pgsql"))
+                              .pgsqlQuery()


Suggested change

.pgsqlQuery()

.query()

The context of pgsql is already clear, no need to repeat.

jfallows · 2024-08-14T22:40:58Z

...o/aklivity/zilla/specs/binding/pgsql/streams/application/create.table/primary.key/client.rpt

+
+read zilla:flush.ext ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .pgsqlCompleted()


Suggested change

.pgsqlCompleted()

.completed()

jfallows · 2024-08-14T22:41:34Z

...o/aklivity/zilla/specs/binding/pgsql/streams/application/create.table/primary.key/client.rpt

+read zilla:flush.ext ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .pgsqlReady()
+                                .status(73)


Let's have an enum for status; IDLE, TRANSACTION, ERROR.

jfallows · 2024-08-14T22:42:31Z

...ts/io/aklivity/zilla/specs/binding/pgsql/streams/network/create.table/primary.key/client.rpt

+      "CREATE TABLE IF NOT EXISTS balances (name VARCHAR, type VARCHAR, description VARCHAR, PRIMARY KEY (name));"
+
+
+read 43                                 # type


Suggested change

read 43 # type

read 43 # type C

jfallows · 2024-08-14T22:42:39Z

...ts/io/aklivity/zilla/specs/binding/pgsql/streams/network/create.table/primary.key/client.rpt

+
+connected
+
+write 51                                # type


Suggested change

write 51 # type

write 51 # type Q

jfallows · 2024-08-14T22:47:18Z

incubator/binding-pgsql.spec/src/main/resources/META-INF/zilla/pgsql.idl

+
+        struct PgsqlReadyFlushEx
+        {
+            int32 status;


enum PgsqlStatus (uint32) { IDLE(73), TRANSACTION(84), ERROR(69) }

Suggested change

int32 status;

PgsqlStatus status;

jfallows · 2024-08-14T22:51:59Z

incubator/binding-pgsql.spec/src/main/resources/META-INF/zilla/pgsql.idl

+            int32 typeOid;
+            int16 length;
+            int16 modifier;
+            int16 text;


enum PgsqlFormat (uint16) { TEXT(0), BINARY(1) }

Suggested change

int16 text;

PgsqlFormat format;

jfallows

Looks good!

Remember to update module-info.java as well.

jfallows · 2024-08-17T01:20:50Z

...src/test/java/io/aklivity/zilla/specs/binding/pgsql/streams/internal/PgsqlFunctionsTest.java

+    }
+
+    @Test
+    public void shouldEncodeAmqpDataExtensionWithDeferred()


jfallows · 2024-08-17T01:21:18Z

...src/test/java/io/aklivity/zilla/specs/binding/pgsql/streams/internal/PgsqlFunctionsTest.java

+        final byte[] array = dataEx()
+            .typeId(0)
+            .query()
+                .build()
+            .build();


Naturally we need to get the test coverage to have a successful build.

jfallows · 2024-08-17T01:22:02Z

...pgsql/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/internal/PgsqlConfiguration.java

+
+public class PgsqlConfiguration extends Configuration
+{
+    private static final ConfigurationDef ECHO_CONFIG;


Suggested change

private static final ConfigurationDef ECHO_CONFIG;

private static final ConfigurationDef PGSQL_CONFIG;

jfallows · 2024-08-17T01:23:22Z

...rc/main/java/io/aklivity/zilla/runtime/binding/pgsql/internal/stream/PgsqlServerFactory.java

+
+    @FunctionalInterface
+    private interface PgsqlServerDecoder
+    {
+        int decode(
+            PgsqlServer server,
+            long traceId,
+            long authorization,
+            long budgetId,
+            DirectBuffer buffer,
+            int offset,
+            int limit);
+    }


Please move to the end of the class.

jfallows · 2024-08-17T01:25:53Z

...rc/main/java/io/aklivity/zilla/runtime/binding/pgsql/internal/stream/PgsqlServerFactory.java

+
+        private void doNetworkData(
+            long traceId,
+            long authorization,
+            int flags,
+            long budgetId,
+            DirectBuffer buffer,
+            int offset,
+            int length)
+        {
+            final int reserved = length + replyPad;
+
+            doData(network, originId, routedId, replyId, replySeq, replyAck, replyMax, traceId, authorization,
+                flags, budgetId, reserved, buffer, offset, length, EMPTY_EXTENSION);
+
+            replySeq += reserved;
+        }


Please move this up with the other doNetworkXxx methods.

jfallows · 2024-08-17T01:28:32Z

...rc/main/java/io/aklivity/zilla/runtime/binding/pgsql/internal/stream/PgsqlServerFactory.java

+
+    private final BufferPool bufferPool;
+    private final MutableDirectBuffer writeBuffer;
+    private final MutableDirectBuffer frameBuffer;


In the PGSQL protocol specification, they use the terminology "message", not "frame".

Let's replace all references to "frame" with "message" here to convey the same concept precisely.

jfallows · 2024-08-22T14:37:28Z

...pgsql.spec/src/test/java/io/aklivity/zilla/specs/binding/pgsql/streams/network/StreamIT.java

@@ -0,0 +1,66 @@
+/*


Please rename to NetworkIT and move up one level.

jfallows · 2024-08-22T14:38:14Z

incubator/binding-pgsql/pom.xml

@@ -24,7 +24,7 @@
  </licenses>

  <properties>
-    <jacoco.coverage.ratio>1.00</jacoco.coverage.ratio>
+    <jacoco.coverage.ratio>0.79</jacoco.coverage.ratio>


This should be above 80% please.

jfallows · 2024-08-22T14:39:11Z

...rc/main/java/io/aklivity/zilla/runtime/binding/pgsql/internal/stream/PgsqlClientFactory.java

+            }
+        }
+
+        private void  doNetworkData(


Suggested change

private void doNetworkData(

private void doNetworkData(

jfallows · 2024-08-22T14:40:54Z

...rc/main/java/io/aklivity/zilla/runtime/binding/pgsql/internal/stream/PgsqlClientFactory.java

+
+        final PgsqlAuthenticationMessageFW pgsqlAuth = authMessageRO.tryWrap(buffer, offset, limit);
+
+        if (pgsqlAuth != null && pgsqlAuth.authenticationType() == 0)


What if a different authentication type is present?
Perhaps we need to reject.

jfallows · 2024-08-22T14:42:46Z

...rc/main/java/io/aklivity/zilla/runtime/binding/pgsql/internal/stream/PgsqlClientFactory.java

+        final PgsqlMessageFW pgsqlType = messageRO.tryWrap(buffer, offset, limit);
+
+        if (pgsqlType != null)
+        {
+            progressOffset = pgsqlType.limit();
+
+            short fieldCount = buffer.getShort(progressOffset, BIG_ENDIAN);
+            progressOffset += Short.BYTES;
+


Does this approach work if pgsql message wraps but the remaining fields following do not?

I think we need to make sure that the progress is only advanced in the returned value if we're moving to a different decoder state, otherwise we would attempt reentry here but not aligned with the decoding logic.

yeah so I thought about this since I don't know if it succeeds to fully decode the message I have below logic at the end of the if condition

if (columns.fieldCount() == fieldCount) { client.onDecodeMessageType(traceId, authorization, columns); client.decoder = decodePgsqlMessage; }

jfallows · 2024-08-22T14:44:20Z

...rc/main/java/io/aklivity/zilla/runtime/binding/pgsql/internal/stream/PgsqlServerFactory.java

+            doNetworkData(traceId, authorization, FLAGS_COMP, 0L, messageBuffer, 0, backendKeyMessage.limit());
+
+            doEncodeParamStatus(traceId, "client_encoding", "UTF8");
+            doEncodeParamStatus(traceId, "standard_confirming_strings", "on");


Suggested change

doEncodeParamStatus(traceId, "standard_confirming_strings", "on");

doEncodeParamStatus(traceId, "standard_conforming_strings", "on");

jfallows · 2024-08-22T14:44:51Z

...rc/main/java/io/aklivity/zilla/runtime/binding/pgsql/internal/stream/PgsqlServerFactory.java

+            doEncodeParamStatus(traceId, "client_encoding", "UTF8");
+            doEncodeParamStatus(traceId, "standard_confirming_strings", "on");
+            doEncodeParamStatus(traceId, "server_version", "1.0.0");
+            doEncodeParamStatus(traceId, "application_name", "psql");


Not zilla?

jfallows · 2024-08-22T14:47:16Z

...rc/main/java/io/aklivity/zilla/runtime/binding/pgsql/internal/stream/PgsqlServerFactory.java

+                if (buffer.getByte(progress.value) == (byte) 0x00)
+                {
+                    progress.addAndGet(Byte.BYTES);
+                }


Why is this part conditional?

this is an indication of the end of a parameter list

jfallows · 2024-08-22T14:51:21Z

...rc/main/java/io/aklivity/zilla/runtime/binding/pgsql/internal/stream/PgsqlServerFactory.java

+
+        return length;
+    }
+


Suggested change

jfallows · 2024-08-22T14:52:23Z

...rc/main/java/io/aklivity/zilla/runtime/binding/pgsql/internal/stream/PgsqlServerFactory.java

+    private int getLengthOfString(
+        DirectBuffer buffer,
+        int startingOffset)
+    {
+        int length = -1;
+
+        length:
+        for (int i = startingOffset; i < buffer.capacity(); i++)
+        {
+            if (buffer.getByte(i) == 0x00)
+            {
+                length = i - startingOffset + 1;
+                break length;
+            }
+        }
+
+        return length;
+    }


Suggested change

private int getLengthOfString(

DirectBuffer buffer,

int startingOffset)

{

int length = -1;

length:

for (int i = startingOffset; i < buffer.capacity(); i++)

{

if (buffer.getByte(i) == 0x00)

{

length = i - startingOffset + 1;

break length;

}

}

return length;

}

private int getLengthOfString(

DirectBuffer buffer,

int offset)

{

int length = -1;

loop:

for (int progress = offset; i < buffer.capacity(); progress++)

{

if (buffer.getByte(progress) == 0x00)

{

length = progress - offset + 1;

break loop;

}

}

return length;

}

jfallows · 2024-08-22T21:21:03Z

...io/aklivity/zilla/specs/binding/pgsql/streams/application/create.table.fragmented/client.rpt

+write zilla:data.ext ${pgsql:dataEx()
+                              .typeId(zilla:id("pgsql"))
+                              .query()
+                                .deferred(41)
+                                .build()
+                              .build()}
+write "CREATE TABLE IF NOT EXISTS balances (name VARCHAR, type VARCHAR, description VARCHAR, timestamp VARCHAR, PRIMARY KEY (name));"
+      [0x00]
+
+write flush


How does this verify fragmented DATA frames?
Don't we need to control the flags explicitly to send 2 different DATA frames, one with INIT flag and deferred in extension, and one with FIN flag plus the remaining deferred payload bytes?

akrambek added 4 commits August 14, 2024 14:29

Update idl

656055c

update schema and add specs

cfcbf38

Add configs

b0b0dae

Update specs

d789aac

jfallows requested changes Aug 14, 2024

View reviewed changes

akrambek added 4 commits August 14, 2024 16:07

Apply feedback

342fb61

Remove extra line

4230164

WIP

83eb9ed

WIP

a63e6db

jfallows previously approved these changes Aug 15, 2024

View reviewed changes

WIP

38fde86

akrambek dismissed jfallows’s stale review via 38fde86 August 15, 2024 01:25

akrambek added 10 commits August 14, 2024 19:34

WIP checkpoint got spect tests passing

9e46eab

WIP

47cc487

WIP

78d9be7

WIP

012bf74

WIP

4e98717

WIP

08cefd7

WIP

bc7390b

WIP

88e8e72

WIP

6e5091b

WIP

71599eb

jfallows reviewed Aug 17, 2024

View reviewed changes

akrambek added 7 commits August 19, 2024 11:12

WIP

2e58cda

WIP

c0d66e4

WIP

4d7fb7d

WIP

e63cb73

WIP

9e7616b

WIP

00ddb6c

WIP

b0199fc

akrambek added 17 commits August 19, 2024 18:48

WIP

c0a9573

WIP

82fdbc4

WIP

0f8e5b6

WIP

58bc184

WIP

bba0e92

WIP

9192ce8

WIP

4f0152c

WIP

8f8a20d

WIP

2c1c584

WIP

04482da

WIP

ae92042

Refactor code

183ddd4

Fix typo

22907f1

Add test coverage

b51a06a

Update coverage

22f5ab3

Move decoder to end of the class

374a09c

Refactor

85c4367

akrambek marked this pull request as ready for review August 22, 2024 06:01

akrambek changed the title ~~Pgsql support~~ Support pgsql binding Aug 22, 2024

jfallows requested changes Aug 22, 2024

View reviewed changes

akrambek added 5 commits August 22, 2024 10:25

Applied feedback from PR

e966aeb

Refactor code

eff671c

WIP

e5b90ec

WIP

97bbb9a

Update coverage

b10833e

jfallows requested changes Aug 22, 2024

View reviewed changes

akrambek added 2 commits August 22, 2024 15:49

Update script for fragmentation

eac30d8

Remove unnecessary setting

3721c45

jfallows approved these changes Aug 23, 2024

View reviewed changes

jfallows merged commit 20230bf into aklivity:develop Aug 23, 2024
5 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support pgsql binding #1200

Support pgsql binding #1200

akrambek commented Aug 14, 2024 •

edited by jfallows

Loading

jfallows Aug 14, 2024

jfallows Aug 14, 2024

jfallows Aug 14, 2024

jfallows Aug 14, 2024

jfallows Aug 14, 2024

jfallows Aug 14, 2024

jfallows Aug 14, 2024

jfallows Aug 14, 2024

jfallows Aug 14, 2024

jfallows Aug 14, 2024

jfallows left a comment

jfallows Aug 17, 2024

jfallows Aug 17, 2024

jfallows Aug 17, 2024

jfallows Aug 17, 2024

jfallows Aug 17, 2024

jfallows Aug 17, 2024

jfallows Aug 22, 2024

jfallows Aug 22, 2024

jfallows Aug 22, 2024

jfallows Aug 22, 2024

jfallows Aug 22, 2024

akrambek Aug 22, 2024

jfallows Aug 22, 2024

jfallows Aug 22, 2024

jfallows Aug 22, 2024

akrambek Aug 22, 2024

jfallows Aug 22, 2024

jfallows Aug 22, 2024

jfallows Aug 22, 2024

		"CREATE TABLE IF NOT EXISTS balances (name VARCHAR, type VARCHAR, description VARCHAR, PRIMARY KEY (name));"


		read 43 # type

	private static final ConfigurationDef ECHO_CONFIG;
	private static final ConfigurationDef PGSQL_CONFIG;


		final PgsqlAuthenticationMessageFW pgsqlAuth = authMessageRO.tryWrap(buffer, offset, limit);

		if (pgsqlAuth != null && pgsqlAuth.authenticationType() == 0)

	doEncodeParamStatus(traceId, "standard_confirming_strings", "on");
	doEncodeParamStatus(traceId, "standard_conforming_strings", "on");

Support pgsql binding #1200

Support pgsql binding #1200

Conversation

akrambek commented Aug 14, 2024 • edited by jfallows Loading

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jfallows 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

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

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

akrambek commented Aug 14, 2024 •

edited by jfallows

Loading