Skip to content

Commit

Permalink
WIP
Browse files Browse the repository at this point in the history
  • Loading branch information
@akrambek
akrambek committed Oct 18, 2024
1 parent 3a9a483 commit 9af3f60
Show file tree
Hide file tree
Showing 13 changed files with 315 additions and 42 deletions.
1 change: 1 addition & 0 deletions ...nding-pgsql/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/parser/PgsqlParser.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ public PgsqlParser()
public String parseCommand(
String sql)
{
commandListener.resetCommand();
parser(sql, commandListener);
return commandListener.command();
}
Expand Down
7 changes: 6 additions & 1 deletion ...main/java/io/aklivity/zilla/runtime/binding/pgsql/parser/listener/SqlCommandListener.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,13 +19,18 @@

public class SqlCommandListener extends PostgreSqlParserBaseListener
{
private String command;
private String command = "";

public String command()
{
return command;
}

public void resetCommand()
{
command = "";
}

@Override
public void enterCreatestmt(
PostgreSqlParser.CreatestmtContext ctx)
Expand Down
13 changes: 6 additions & 7 deletions ...inding/risingwave/streams/effective/create.table.with.primary.key.and.includes/client.rpt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -67,11 +67,10 @@ write zilla:data.ext ${pgsql:dataEx()
.query()
.build()
.build()}
write "CREATE MATERIALIZED VIEW IF NOT EXISTS cities_view AS"
" SELECT id, name, description,"
" COALESCE(correlation_id, zilla_correlation_id_header::varchar) as correlation_id,"
" COALESCE(identity, zilla_identity_header::varchar) as identity,"
" timestamp"
write "CREATE MATERIALIZED VIEW IF NOT EXISTS cities_view"
" AS SELECT id, name, description,"
" COALESCE(zilla_correlation_id, zilla_correlation_id_header::varchar) as zilla_correlation_id,"
" COALESCE(zilla_identity, zilla_identity_header::varchar) as zilla_identity, timestamp"
" FROM cities_source;"
[0x00]
write flush
Expand All @@ -97,7 +96,7 @@ write zilla:data.ext ${pgsql:dataEx()
.build()}
write "CREATE TABLE IF NOT EXISTS cities"
" (id VARCHAR, name VARCHAR, description VARCHAR,"
" correlation_id VARCHAR, identity VARCHAR, timestamp TIMESTAMP,"
" zilla_correlation_id VARCHAR, zilla_identity VARCHAR, timestamp TIMESTAMP,"
" PRIMARY KEY (id));"
[0x00]
write flush
Expand Down Expand Up @@ -192,7 +191,7 @@ write zilla:data.ext ${pgsql:dataEx()
.build()}
write "CREATE TOPIC IF NOT EXISTS cities "
"(id VARCHAR, name VARCHAR, description VARCHAR,"
" correlation_id VARCHAR, identity VARCHAR, timestamp TIMESTAMP,"
" zilla_correlation_id VARCHAR, zilla_identity VARCHAR, timestamp TIMESTAMP,"
" PRIMARY KEY (id));"
[0x00]
write flush
Expand Down
13 changes: 6 additions & 7 deletions ...inding/risingwave/streams/effective/create.table.with.primary.key.and.includes/server.rpt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -70,11 +70,10 @@ read zilla:data.ext ${pgsql:dataEx()
.query()
.build()
.build()}
read "CREATE MATERIALIZED VIEW IF NOT EXISTS cities_view AS"
" SELECT id, name, description,"
" COALESCE(correlation_id, zilla_correlation_id_header::varchar) as correlation_id,"
" COALESCE(identity, zilla_identity_header::varchar) as identity,"
" timestamp"
read "CREATE MATERIALIZED VIEW IF NOT EXISTS cities_view"
" AS SELECT id, name, description,"
" COALESCE(zilla_correlation_id, zilla_correlation_id_header::varchar) as zilla_correlation_id,"
" COALESCE(zilla_identity, zilla_identity_header::varchar) as zilla_identity, timestamp"
" FROM cities_source;"
[0x00]

Expand All @@ -99,7 +98,7 @@ read zilla:data.ext ${pgsql:dataEx()
.build()}
read "CREATE TABLE IF NOT EXISTS cities"
" (id VARCHAR, name VARCHAR, description VARCHAR,"
" correlation_id VARCHAR, identity VARCHAR, timestamp TIMESTAMP,"
" zilla_correlation_id VARCHAR, zilla_identity VARCHAR, timestamp TIMESTAMP,"
" PRIMARY KEY (id));"
[0x00]

Expand Down Expand Up @@ -193,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 VARCHAR, timestamp TIMESTAMP,"
" zilla_correlation_id VARCHAR, zilla_identity VARCHAR, timestamp TIMESTAMP,"
" PRIMARY KEY (id));"
[0x00]

Expand Down
7 changes: 3 additions & 4 deletions ...cs/binding/risingwave/streams/pgsql/create.table.with.primary.key.and.includes/client.rpt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,10 +33,9 @@ write zilla:data.ext ${pgsql:dataEx()
.build()
.build()}
write "CREATE TABLE IF NOT EXISTS cities "
"(id VARCHAR, name VARCHAR, description VARCHAR, PRIMARY KEY (id))\n"
"INCLUDE zilla_correlation_id AS correlation_id\n"
"INCLUDE zilla_identity AS identity\n"
"INCLUDE timestamp as timestamp;"
"(id VARCHAR, name VARCHAR, description VARCHAR,"
" zilla_correlation_id VARCHAR, zilla_identity VARCHAR, timestamp TIMESTAMP,"
" PRIMARY KEY (id));"
[0x00]

write flush
Expand Down
7 changes: 3 additions & 4 deletions ...cs/binding/risingwave/streams/pgsql/create.table.with.primary.key.and.includes/server.rpt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -37,10 +37,9 @@ read zilla:data.ext ${pgsql:dataEx()
.build()
.build()}
read "CREATE TABLE IF NOT EXISTS cities "
"(id VARCHAR, name VARCHAR, description VARCHAR, PRIMARY KEY (id))\n"
"INCLUDE zilla_correlation_id AS correlation_id\n"
"INCLUDE zilla_identity AS identity\n"
"INCLUDE timestamp as timestamp;"
"(id VARCHAR, name VARCHAR, description VARCHAR,"
" zilla_correlation_id VARCHAR, zilla_identity VARCHAR, timestamp TIMESTAMP,"
" PRIMARY KEY (id));"
[0x00]

write advise zilla:flush ${pgsql:flushEx()
Expand Down
12 changes: 7 additions & 5 deletions ...ntime/binding/risingwave/internal/statement/RisingwaveCreateMaterializedViewTemplate.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
*/
package io.aklivity.zilla.runtime.binding.risingwave.internal.statement;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.stream.Collectors;

Expand All @@ -26,7 +27,7 @@ public class RisingwaveCreateMaterializedViewTemplate extends RisingwaveCommandT
private final String sqlFormat = """
CREATE MATERIALIZED VIEW IF NOT EXISTS %s AS %s;\u0000""";
private final String fieldFormat = "%s, ";
private final String includeFormat = "COALESCE(%s, zilla_%s_header::varchar) as %s, ";
private final String includeFormat = "COALESCE(%s, %s_header::varchar) as %s, ";

public RisingwaveCreateMaterializedViewTemplate()
{
Expand All @@ -50,25 +51,26 @@ public String generate(

Map<String, String> includes = tableInfo.columns().entrySet().stream()
.filter(e -> ZILLA_MAPPINGS.containsKey(e.getKey()))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
.collect(LinkedHashMap::new, (m, e) -> m.put(e.getKey(), e.getValue()), Map::putAll);

if (!includes.isEmpty())
{
fieldBuilder.setLength(0);

tableInfo.columns().values().stream()
tableInfo.columns().keySet()
.stream()
.filter(c -> !ZILLA_MAPPINGS.containsKey(c))
.forEach(c -> fieldBuilder.append(String.format(fieldFormat, c)));

includes.forEach((k, v) ->
{
if ("timestamp".equals(k))
{
fieldBuilder.append(String.format(fieldFormat, v));
fieldBuilder.append(String.format(fieldFormat, k));
}
else
{
fieldBuilder.append(String.format(includeFormat, v, v, v));
fieldBuilder.append(String.format(includeFormat, k, k, k));
}
});

Expand Down
7 changes: 4 additions & 3 deletions ...y/zilla/runtime/binding/risingwave/internal/statement/RisingwaveCreateSourceTemplate.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
*/
package io.aklivity.zilla.runtime.binding.risingwave.internal.statement;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.stream.Collectors;

Expand Down Expand Up @@ -78,7 +79,7 @@ public String generateTableSource(
includeBuilder.setLength(0);
Map<String, String> includes = tableInfo.columns().entrySet().stream()
.filter(e -> ZILLA_MAPPINGS.containsKey(e.getKey()))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
.collect(LinkedHashMap::new, (m, e) -> m.put(e.getKey(), e.getValue()), Map::putAll);

if (!includes.isEmpty())
{
Expand All @@ -87,11 +88,11 @@ public String generateTableSource(
{
if ("timestamp".equals(k))
{
includeBuilder.append(String.format(ZILLA_MAPPINGS.get(k), v));
includeBuilder.append(String.format(ZILLA_MAPPINGS.get(k), k));
}
else
{
includeBuilder.append(String.format(ZILLA_MAPPINGS.get(k), "zilla_%s_header".formatted(v)));
includeBuilder.append(String.format(ZILLA_MAPPINGS.get(k), "%s_header".formatted(k)));
}
});
includeBuilder.delete(includeBuilder.length() - 1, includeBuilder.length());
Expand Down
10 changes: 0 additions & 10 deletions ...ty/zilla/runtime/binding/risingwave/internal/statement/RisingwaveCreateTableTemplate.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,6 @@
*/
package io.aklivity.zilla.runtime.binding.risingwave.internal.statement;

import java.util.Map;
import java.util.stream.Collectors;

import io.aklivity.zilla.runtime.binding.pgsql.parser.module.TableInfo;

public class RisingwaveCreateTableTemplate extends RisingwaveCommandTemplate
Expand Down Expand Up @@ -44,13 +41,6 @@ public String generate(
.forEach((k, v) -> fieldBuilder.append(
String.format(fieldFormat, k, v)));

Map<String, String> includes = tableInfo.columns().entrySet().stream()
.filter(e -> ZILLA_MAPPINGS.containsKey(e.getKey()))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

includes.forEach((k, v) -> fieldBuilder.append(
String.format(fieldFormat, v, ZILLA_INCLUDE_TYPE_MAPPINGS.get(k))));

fieldBuilder.delete(fieldBuilder.length() - 2, fieldBuilder.length());

return String.format(sqlFormat, topic, fieldBuilder, primaryKey);
Expand Down
2 changes: 1 addition & 1 deletion .../io/aklivity/zilla/runtime/binding/risingwave/internal/stream/RisingwaveProxyFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@

public final class RisingwaveProxyFactory implements RisingwaveStreamFactory
{
private static final String SPLIT_STATEMENTS = "\"(?<=;)(?!\\x00)\"";
private static final String SPLIT_STATEMENTS = "(?<=;)(?!\\x00)";
private static final int END_OF_FIELD = 0x00;

private static final int FLAGS_INIT = 0x02;
Expand Down
78 changes: 78 additions & 0 deletions ...e/binding/risingwave/internal/statement/RisingwaveCreateMaterializedViewTemplateTest.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
package io.aklivity.zilla.runtime.binding.risingwave.internal.statement;

import static org.junit.Assert.assertEquals;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;

import org.junit.Ignore;
import org.junit.Test;

import io.aklivity.zilla.runtime.binding.pgsql.parser.module.TableInfo;
import io.aklivity.zilla.runtime.binding.pgsql.parser.module.ViewInfo;

public class RisingwaveCreateMaterializedViewTemplateTest
{
private final RisingwaveCreateMaterializedViewTemplate template = new RisingwaveCreateMaterializedViewTemplate();

@Test
public void shouldGenerateMaterializedViewWithValidViewInfo()
{
ViewInfo viewInfo = new ViewInfo("test_view", "SELECT * FROM test_table");
String expectedSQL = """
CREATE MATERIALIZED VIEW IF NOT EXISTS test_view AS SELECT * FROM test_table;\u0000""";

String actualSQL = template.generate(viewInfo);

assertEquals(expectedSQL, actualSQL);
}

@Ignore("TODO")
@Test
public void shouldGenerateMaterializedViewWithValidTableInfo()
{
TableInfo tableInfo = new TableInfo(
"test_table",
Map.of("id", "INT", "name", "STRING"),
Set.of("id"));
String expectedSQL = """
CREATE MATERIALIZED VIEW IF NOT EXISTS test_table_view AS SELECT id, name FROM test_table_source;\u0000""";

String actualSQL = template.generate(tableInfo);

assertEquals(expectedSQL, actualSQL);
}

@Test
public void shouldGenerateMaterializedViewWithEmptyColumns()
{
TableInfo tableInfo = new TableInfo("empty_table", Map.of(), Set.of());
String expectedSQL = """
CREATE MATERIALIZED VIEW IF NOT EXISTS empty_table_view AS SELECT * FROM empty_table_source;\u0000""";

String actualSQL = template.generate(tableInfo);

assertEquals(expectedSQL, actualSQL);
}

@Test
public void shouldGenerateMaterializedViewWithIncludes()
{
Map<String, String> columns = new LinkedHashMap<>();
columns.put("id", "INT");
columns.put("zilla_correlation_id", "VARCHAR");
columns.put("zilla_identity", "VARCHAR");
columns.put("timestamp", "TIMESTAMP");

TableInfo tableInfo = new TableInfo("test_table", columns, Set.of("id"));
String expectedSQL = "CREATE MATERIALIZED VIEW IF NOT EXISTS test_table_view AS SELECT id," +
" COALESCE(zilla_correlation_id, zilla_correlation_id_header::varchar) as zilla_correlation_id," +
" COALESCE(zilla_identity, zilla_identity_header::varchar) as zilla_identity, timestamp" +
" FROM test_table_source;\u0000";

String actualSQL = template.generate(tableInfo);

assertEquals(expectedSQL, actualSQL);
}
}
Loading

0 comments on commit 9af3f60

Please sign in to comment.