Support ZVIEW command #1329

akrambek · 2024-11-14T00:28:15Z

Description

Support Zview statement

jfallows · 2024-11-14T00:54:51Z

...ing-pgsql/src/main/antlr4/io/aklivity/zilla/runtime/binding/pgsql/parser/PostgreSqlParser.g4

+    | ZSTREAM
+    | ZVIEW
+    | ZTABLE


jfallows · 2024-11-14T00:56:35Z

...klivity/zilla/specs/binding/risingwave/streams/pgsql/create.zstream.with.includes/client.rpt


+write "CREATE ZSTREAM weather (\n"


Let's call this commands instead so it fits better to have reply-to replies, and structure the columns accordingly.

jfallows · 2024-11-14T00:58:05Z

...klivity/zilla/specs/binding/risingwave/streams/pgsql/create.zstream.with.includes/client.rpt

+      "    reply_to = 'replies',\n"
+      "    function = compute_status_and_correlation_id,\n"


We need to create these too, so that the example is self-documenting and self-consistent, agree?

Also, after this first pass, we wanted the function to become less aware of correlation_id, and handle that implicitly, agreed?

How do we route specific commands to specific functions?
This seems to currently imply every function has its own ZSTREAM and therefore it's own pair of command and reply topics.

jfallows

How many root pgsql connections are created to risingwave per regular user pgsql connection to zilla?

jfallows · 2024-11-27T06:34:11Z

...ding-pgsql/src/test/java/io/aklivity/zilla/runtime/binding/pgsql/parser/PgsqlParserTest.java

@@ -207,17 +207,17 @@ public void shouldHandleEmptyDropStatement()
    @Test
    public void shouldParseDropView()
    {
-        String sql = "DROP VIEW test_view;";
+        String sql = "DROP ZVIEW test_view;";


Should this change be reverted since this is shouldParseDropView() and we also have a separate shouldParseDropZView() ?

jfallows · 2024-11-27T06:35:14Z

...aklivity/zilla/specs/binding/risingwave/streams/effective/alter.stream.add.column/server.rpt

@@ -19,6 +19,18 @@ accept "zilla://streams/app1"

 accepted

+read zilla:begin.ext ${pgsql:beginEx()


Suggested change

read zilla:begin.ext ${pgsql:beginEx()

read zilla:begin.ext ${pgsql:matchBeginEx()

Same feedback for other scripts, where we are reading the pgsql begin extension.

jfallows · 2024-11-27T06:47:01Z

...ava/io/aklivity/zilla/runtime/binding/risingwave/internal/stream/RisingwaveProxyFactory.java

+            binding.routes.forEach(r ->
+            {
+                final long clientId = r.id + DEFAULT_USER_HASH;
+                streamsByRouteIds.put(clientId,
+                    new PgsqlClient(this, routedId, r.id, clientId, DEFAULT_USER));
+            });
+
+            final long clientId = risingwaveRouteId + ZILLABASE_USER_HASH;
+            streamsByRouteIds.put(clientId,
+                new PgsqlClient(this, routedId, risingwaveRouteId, clientId, ZILLABASE_USER));
+


We should discuss the design problem you are trying to solve with this approach.

I think we discussed changing this, where zillabase is the system user that could potentially be done just once (per core). Let me know if there are any changes you'd like to make to prepare for that and we can file an enhancement to track the optimization for after the feature work is completed.

akrambek · 2024-11-27T07:38:43Z

One

jfallows · 2024-12-09T15:19:02Z

...ava/io/aklivity/zilla/runtime/binding/pgsql/parser/listener/SqlAlterStreamTopicListener.java

@@ -29,6 +29,7 @@ public class SqlAlterStreamTopicListener extends PostgreSqlParserBaseListener
 {
    private final TokenStream tokens;

+    private String schema;
    private String name;
    private final List<AlterExpression> alterExpressions = new ArrayList<>();


Please move final field up with other final fields and initialize in constructor as it is not a (pseudo-) constant.

jfallows · 2024-12-09T15:19:34Z

...java/io/aklivity/zilla/runtime/binding/pgsql/parser/listener/SqlAlterTableTopicListener.java

@@ -29,6 +29,7 @@ public class SqlAlterTableTopicListener extends PostgreSqlParserBaseListener
 {
    private final TokenStream tokens;

+    private String schema;
    private String name;
    private final List<AlterExpression> alterExpressions = new ArrayList<>();


Same feedback.

jfallows · 2024-12-09T15:20:16Z

...in/java/io/aklivity/zilla/runtime/binding/pgsql/parser/listener/SqlCreateStreamListener.java

-        name = ctx.getText();
+        String text = ctx.getText();
+        String[] split = text.split("\\.");
+        schema = split.length > 1 ? split[0] : "public";


Please move this default "public" to a constant.

jfallows · 2024-12-09T15:20:29Z

...ava/io/aklivity/zilla/runtime/binding/pgsql/parser/listener/SqlCreateTableTopicListener.java

-        name = ctx.getText();
+        String text = ctx.getText();
+        String[] split = text.split("\\.");
+        schema = split.length > 1 ? split[0] : "public";


Same feedback.

jfallows · 2024-12-09T15:20:42Z

...ain/java/io/aklivity/zilla/runtime/binding/pgsql/parser/listener/SqlCreateZviewListener.java

-        name = ctx.create_mv_target().qualified_name().getText();
+        String text = ctx.create_mv_target().qualified_name().getText();
+        String[] split = text.split("\\.");
+        schema = split.length > 1 ? split[0] : "public";


Same feedback.

jfallows · 2024-12-09T15:20:52Z

...l/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/parser/listener/SqlDropListener.java

+        {
+            String text = n.getText();
+            String[] split = text.split("\\.");
+            String schema = split.length > 1 ? split[0] : "public";


Same feedback.

jfallows · 2024-12-09T15:22:15Z

...ding-pgsql/src/test/java/io/aklivity/zilla/runtime/binding/pgsql/parser/PgsqlParserTest.java

@@ -52,7 +53,7 @@ public void shouldParseWithPrimaryKeySql()
    }

    @Test
-    public void shouldCreateParseWithPrimaryKeysSql()
+    public void shouldCreateTableParseWithPrimaryKeysSql()


Suggested change

public void shouldCreateTableParseWithPrimaryKeysSql()

public void shouldCreateTableWithPrimaryKey()

works?

Note: we are in the parser test class, so no need to use parse in the name of any of the test methods, please review all of the test method names.

jfallows · 2024-12-09T15:24:03Z

...ding-pgsql/src/test/java/io/aklivity/zilla/runtime/binding/pgsql/parser/PgsqlParserTest.java

+        assertEquals("public", drops.get(0).schema());
+        assertEquals("table1", drops.get(0).name());
+        assertEquals("public", drops.get(1).schema());
+        assertEquals("table2", drops.get(1).name());
    }

    @Test(expected = ParseCancellationException.class)
    public void shouldHandleEmptyDropStatement()


Suggested change

public void shouldHandleEmptyDropStatement()

public void shouldDropTable()

jfallows · 2024-12-09T15:33:35Z

...ava/io/aklivity/zilla/runtime/binding/risingwave/internal/stream/RisingwaveProxyFactory.java

+            binding.routes.forEach(r ->
+            {
+                final long clientId = r.id + DEFAULT_USER_HASH;
+                streamsByRouteIds.put(clientId,
+                    new PgsqlClient(this, routedId, r.id, clientId, DEFAULT_USER));
+            });
+
+            final long clientId = risingwaveRouteId + ZILLABASE_USER_HASH;
+            streamsByRouteIds.put(clientId,
+                new PgsqlClient(this, routedId, risingwaveRouteId, clientId, ZILLABASE_USER));
+


I think we discussed changing this, where zillabase is the system user that could potentially be done just once (per core). Let me know if there are any changes you'd like to make to prepare for that and we can file an enhancement to track the optimization for after the feature work is completed.

jfallows · 2024-12-10T01:48:03Z

...ivity/zilla/runtime/binding/pgsql/kafka/internal/schema/PgsqlKafkaKeyAvroSchemaTemplate.java

@@ -31,15 +31,15 @@ public PgsqlKafkaKeyAvroSchemaTemplate(

    public String generate(
        String database,
-        Table table)
+        CreateTable createTable)


Suggested change

CreateTable createTable)

CreateTable command)

Seems more readable, agreed?

Let's update the other generate methods too, so it's all consistent.

jfallows · 2024-12-10T18:30:25Z

.../binding-pgsql/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/parser/model/Alter.java

@@ -17,6 +17,7 @@
 import java.util.List;

 public record Alter(


AlterZView AlterZTable ?

we don't support alter zview and ztable is addressed in Ztable PR

jfallows · 2024-12-10T18:30:35Z

...g-pgsql/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/parser/model/CreateStream.java

@@ -16,7 +16,8 @@

 import java.util.Map;

-public record Stream(
+public record CreateStream(


CreateZStream?

it will be in another PR

jfallows · 2024-12-11T17:59:31Z

...ng-pgsql/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/parser/model/CreateTable.java

@@ -17,7 +17,8 @@
 import java.util.List;
 import java.util.Set;

-public record Table(
+public record CreateTable(


CreateZTable

?

it is covered in ZTABLE PR

akrambek added 3 commits November 12, 2024 15:01

Update antlr gramma

7f4b2da

Refactor scripts for zstream

84eacb3

Remove redundant files

a5f3121

jfallows reviewed Nov 14, 2024

View reviewed changes

akrambek added 18 commits November 13, 2024 17:50

Remove redunant files

cfe8ade

Fix typo

519279b

Replace weather with commands

b60ee1e

Remove redundant exclude

912bd26

WIP

1085600

Merge branch 'develop' into feature/z-catalog

7bc5c5d

Revert back the change

4039029

Checkpoint zview

8a5ace9

Stable state

92930f8

WIP

463d426

WIP

bed9c1c

WIP

586ea67

WIP

ac7758d

WIP

8372f5c

WIP

3dc8c3a

WIP

478be8f

WIP

d1b247c

Checkpoint

546473c

akrambek marked this pull request as ready for review November 27, 2024 00:59

Remove uncessary files

99b7216

akrambek changed the title ~~Update schemas to catalog z related queries into a separate schema and make public schema as default~~ Support ZVIEW command Nov 27, 2024

jfallows reviewed Nov 27, 2024

View reviewed changes

akrambek added 3 commits November 27, 2024 15:26

WIP

f6435a3

WIP

4bbc4aa

WIP

8971b87

akrambek added 11 commits November 28, 2024 12:43

WIP

9dee812

WIP

cc6c80e

Change user

212872d

Checkpoint

21e6963

Update the script

e4377a0

WIP

b052c58

WIP

5401488

WIP

7fc64be

WIP

5a286cc

WIP

38f7371

Checkpoint

e8f5e7e

jfallows requested changes Dec 9, 2024

View reviewed changes

akrambek added 5 commits December 9, 2024 10:59

WIP

a45c87c

WIP

8a9305b

Fix remaining issues

cd63668

WIP

93126b7

Fix stylecheck errors

17fdc27

jfallows requested changes Dec 11, 2024

View reviewed changes

akrambek added 2 commits December 11, 2024 10:58

Applied feedback

d5764c2

Remove redundant file

4941ad0

jfallows approved these changes Dec 13, 2024

View reviewed changes

jfallows merged commit bbbc39e into aklivity:develop Dec 13, 2024
5 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support ZVIEW command #1329

Support ZVIEW command #1329

akrambek commented Nov 14, 2024 •

edited

Loading

jfallows Nov 14, 2024

jfallows Nov 14, 2024

jfallows Nov 14, 2024

jfallows left a comment

jfallows Nov 27, 2024

akrambek Nov 27, 2024

jfallows Nov 27, 2024

jfallows Nov 27, 2024

jfallows Dec 9, 2024

akrambek commented Nov 27, 2024

jfallows Dec 9, 2024

jfallows Dec 9, 2024

jfallows Dec 9, 2024

jfallows Dec 9, 2024

jfallows Dec 9, 2024

jfallows Dec 9, 2024

jfallows Dec 9, 2024

jfallows Dec 9, 2024

jfallows Dec 9, 2024

jfallows Dec 10, 2024

jfallows Dec 10, 2024

akrambek Dec 11, 2024

jfallows Dec 10, 2024

akrambek Dec 11, 2024

jfallows Dec 11, 2024

akrambek Dec 11, 2024

		" reply_to = 'replies',\n"
		" function = compute_status_and_correlation_id,\n"

		@@ -19,6 +19,18 @@ accept "zilla://streams/app1"

		accepted

		read zilla:begin.ext ${pgsql:beginEx()

	read zilla:begin.ext ${pgsql:beginEx()
	read zilla:begin.ext ${pgsql:matchBeginEx()

	public void shouldCreateTableParseWithPrimaryKeysSql()
	public void shouldCreateTableWithPrimaryKey()

	public void shouldHandleEmptyDropStatement()
	public void shouldDropTable()

Support ZVIEW command #1329

Support ZVIEW command #1329

Conversation

akrambek commented Nov 14, 2024 • edited Loading

Description

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

akrambek commented Nov 27, 2024

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 Nov 14, 2024 •

edited

Loading