Skip to content

Commit 6d27256

Browse files
mihaibudiumihaibudiu
committed
[SQL] Compiler support for new materialized view Catalog API
Signed-off-by: Mihai Budiu <[email protected]>
1 parent 04e58fe commit 6d27256

8 files changed

Lines changed: 223 additions & 81 deletions

File tree

‎sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPViewBaseOperator.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
public abstract class DBSPViewBaseOperator extends DBSPUnaryOperator {
1919
// Called viewName, but it could also be an index name
2020
public final ProgramIdentifier viewName;
21+
/** For a view holds the query, for an index holds the view name */
2122
public final String query;
2223
/** Usually a struct, but can be a tuple for indexes */
2324
public final DBSPType originalRowType;

‎sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java‎

Lines changed: 168 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,7 @@
8989
import org.dbsp.sqlCompiler.compiler.errors.UnimplementedException;
9090
import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.ProgramIdentifier;
9191
import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject;
92+
import org.dbsp.sqlCompiler.compiler.frontend.parser.SqlCreateView;
9293
import org.dbsp.sqlCompiler.compiler.frontend.statements.IHasSchema;
9394
import org.dbsp.sqlCompiler.compiler.visitors.VisitDecision;
9495
import org.dbsp.sqlCompiler.compiler.visitors.inner.CanonicalForm;
@@ -134,10 +135,13 @@
134135

135136
import javax.annotation.Nullable;
136137
import java.util.ArrayList;
138+
import java.util.HashMap;
137139
import java.util.HashSet;
138140
import java.util.Iterator;
139141
import java.util.List;
140142
import java.util.Locale;
143+
import java.util.Map;
144+
import java.util.Objects;
141145
import java.util.Set;
142146

143147
/** This visitor generates a Rust implementation of a circuit. */
@@ -607,14 +611,14 @@ void registerTable(DBSPSourceMultisetOperator operator, DBSPOperator materialize
607611
j = this.stripConnectors(j);
608612
DBSPStrLiteral json = new DBSPStrLiteral(j.toString(), true);
609613
operator.originalRowType.accept(this.innerVisitor);
610-
this.builder.append(">(")
614+
this.builder.append(">(").increase()
611615
.append(materialized.getOutput(0).getName(this.preferHash))
612-
.append(".clone(), ")
616+
.append(".clone(),").newline()
613617
.append(this.handleName(operator))
614-
.append(".clone(), ");
618+
.append(".clone(),").newline();
615619
json.accept(this.innerVisitor);
616620
this.innerVisitor.setOperatorContext(null);
617-
this.builder.append(");");
621+
this.builder.newline().decrease().append(");");
618622
}
619623

620624
@Override
@@ -832,23 +836,23 @@ public VisitDecision sourceMapPostfix(IInputMapOperator operator) {
832836
upsertStruct.toTupleDeep().accept(this.innerVisitor);
833837
this.builder.append(", ");
834838
upsertStruct.accept(this.innerVisitor);
835-
this.builder.append(", _, _>(")
839+
this.builder.append(", _, _>(").increase()
836840
.append(operator.asOperator().getOutput(operator.getDataOutputIndex()).getName(this.preferHash))
837-
.append(".clone(), ")
841+
.append(".clone(),").newline()
838842
.append(this.handleName(operator.asOperator()))
839-
.append(".clone() , ");
843+
.append(".clone(),").newline();
840844

841845
CanonicalForm cf = new CanonicalForm(this.compiler);
842846
DBSPExpression key = operator.getKeyFunc();
843847
key = cf.apply(key).to(DBSPExpression.class);
844848
key.accept(this.innerVisitor);
845-
this.builder.append(", ");
849+
this.builder.append(",").newline();
846850
DBSPExpression update = operator.getUpdateKeyFunc(upsertStruct);
847851
update = cf.apply(update).to(DBSPExpression.class);
848852
update.accept(this.innerVisitor);
849-
this.builder.append(", ");
853+
this.builder.append(",").newline();
850854
json.accept(this.innerVisitor);
851-
this.builder.append(");")
855+
this.builder.newline().decrease().append(");")
852856
.newline();
853857
}
854858
this.innerVisitor.setOperatorContext(null);
@@ -1042,20 +1046,58 @@ public VisitDecision preorder(DBSPWaterlineOperator operator) {
10421046
return VisitDecision.STOP;
10431047
}
10441048

1049+
private void appendIndexKeys(DBSPSinkOperator operator) {
1050+
Utilities.enforce(operator.isIndex());
1051+
DBSPTypeRawTuple raw = operator.originalRowType.to(DBSPTypeRawTuple.class);
1052+
Utilities.enforce(raw.size() == 2);
1053+
DBSPTypeStruct keyStruct = raw.tupFields[0].to(DBSPTypeStruct.class);
1054+
boolean first = true;
1055+
this.builder.append("&[");
1056+
for (Iterator<ProgramIdentifier> it = keyStruct.getFieldNames(); it.hasNext(); ) {
1057+
ProgramIdentifier field = it.next();
1058+
if (!first)
1059+
this.builder.append(",");
1060+
first = false;
1061+
this.builder.append("\"")
1062+
.append(field.toString())
1063+
.append("\".to_string()");
1064+
}
1065+
this.builder.append("]");
1066+
}
1067+
10451068
@Override
10461069
public VisitDecision preorder(DBSPSinkOperator operator) {
1047-
this.writeComments(operator);
10481070
this.innerVisitor.setOperatorContext(operator);
10491071
if (!this.useHandles) {
1072+
/*
1073+
Current register API:
1074+
- `register_materialized_output_zset_persistent` (existing method) - to be used to register
1075+
materialized views _without_ indexes only.
1076+
- `register_materialized_output_map_persistent` - used to simultaneously register a
1077+
materialized view and an index, which will share a common integral.
1078+
- `register_materialized_index_persistent` - used with views that have multiple indexes
1079+
to register the second, third, etc. index.
1080+
*/
1081+
10501082
final DBSPType type = operator.originalRowType;
10511083
this.generateNestedStructs(type, false);
10521084
this.computeHash(operator);
10531085
if (operator.isIndex()) {
1086+
ViewAndIndexes indexes = this.indexes.get(operator.query);
1087+
Utilities.enforce(indexes != null);
1088+
boolean registerAsMap =
1089+
indexes.getView().metadata.viewKind == SqlCreateView.ViewKind.MATERIALIZED
1090+
&& indexes.firstIndex() == operator;
1091+
String registerFunction = registerAsMap
1092+
? "register_materialized_output_map_persistent"
1093+
: "register_materialized_index_persistent";
1094+
this.writeComments(operator);
10541095
DBSPTypeRawTuple raw = operator.originalRowType.to(DBSPTypeRawTuple.class);
10551096
Utilities.enforce(raw.size() == 2);
10561097

1057-
this.builder.append("catalog.register_index_persistent");
1058-
this.builder.append("::<").increase();
1098+
this.builder.append("catalog.")
1099+
.append(registerFunction)
1100+
.append("::<").increase();
10591101
operator.getOutputIndexedZSetType().keyType.accept(this.innerVisitor);
10601102
this.builder.append(",").newline();
10611103
raw.tupFields[0].accept(this.innerVisitor);
@@ -1064,49 +1106,61 @@ public VisitDecision preorder(DBSPSinkOperator operator) {
10641106
this.builder.append(",").newline();
10651107
raw.tupFields[1].accept(this.innerVisitor);
10661108
this.builder.decrease().append(">");
1067-
this.builder.append("(hash, ").newline()
1109+
this.builder.append("(").increase().append("hash,").newline()
10681110
.append(this.getInputName(operator, 0))
1069-
.append(".clone()")
1070-
.append(", &SqlIdentifier::from(\"");
1071-
this.builder.append(operator.viewName.toString())
1072-
.append("\"), &SqlIdentifier::from(\"")
1073-
.append(operator.query);
1074-
this.builder.append("\"),")
1075-
.newline().append("&vec!(");
1076-
DBSPTypeStruct keyStruct = raw.tupFields[0].to(DBSPTypeStruct.class);
1077-
boolean first = true;
1078-
for (Iterator<ProgramIdentifier> it = keyStruct.getFieldNames(); it.hasNext(); ) {
1079-
ProgramIdentifier field = it.next();
1080-
if (!first)
1081-
this.builder.append(",");
1082-
first = false;
1083-
this.builder.append("&SqlIdentifier::from(\"")
1084-
.append(field.toString())
1085-
.append("\")");
1111+
.append(".clone(),").newline();
1112+
if (registerAsMap) {
1113+
this.builder.append("Some(SqlIdentifier::from(\"");
1114+
this.builder.append(operator.viewName.toString())
1115+
.append("\")),").newline();
1116+
// schema
1117+
IHasSchema description = this.metadata.getViewDescription(indexes.getView().viewName);
1118+
JsonNode j = description.asJson(true);
1119+
j = this.stripConnectors(j);
1120+
DBSPStrLiteral schemaJson = new DBSPStrLiteral(j.toString(), true);
1121+
schemaJson.accept(this.innerVisitor);
1122+
this.builder.append(",").newline();
1123+
} else {
1124+
this.builder.append("&SqlIdentifier::from(\"");
1125+
this.builder.append(operator.viewName.toString())
1126+
.append("\"),").newline();
1127+
this.builder.append("& SqlIdentifier::from(\"")
1128+
.append(operator.query)
1129+
.append("\"),")
1130+
.newline();
10861131
}
1087-
this.builder.append("));")
1132+
this.appendIndexKeys(operator);
1133+
this.builder.decrease().newline().append(");")
10881134
.newline();
10891135
} else {
1090-
IHasSchema description = this.metadata.getViewDescription(operator.viewName);
1091-
JsonNode j = description.asJson(true);
1092-
j = this.stripConnectors(j);
1093-
DBSPStrLiteral json = new DBSPStrLiteral(j.toString(), true);
1094-
String registerFunction = switch (operator.metadata.viewKind) {
1095-
case MATERIALIZED -> "register_materialized_output_zset_persistent";
1096-
case LOCAL -> throw new InternalCompilerError("Sink operator for local view " + operator);
1097-
case STANDARD -> "register_output_zset_persistent";
1098-
};
1099-
this.builder.append("catalog.")
1100-
.append(registerFunction)
1101-
.append("::<_, ");
1102-
operator.originalRowType.accept(this.innerVisitor);
1103-
this.builder.append(">(hash, ")
1104-
.append(this.getInputName(operator, 0))
1105-
.append(".clone()")
1106-
.append(", ");
1107-
json.accept(this.innerVisitor);
1108-
this.builder.append(");")
1109-
.newline();
1136+
// Regular view
1137+
ViewAndIndexes indexes = this.indexes.get(operator.viewName.name());
1138+
if (indexes.isEmpty() || operator.metadata.viewKind != SqlCreateView.ViewKind.MATERIALIZED) {
1139+
// Materialized views with indexes are not emitted at all
1140+
this.writeComments(operator);
1141+
IHasSchema description = this.metadata.getViewDescription(operator.viewName);
1142+
JsonNode j = description.asJson(true);
1143+
j = this.stripConnectors(j);
1144+
DBSPStrLiteral schemaJson = new DBSPStrLiteral(j.toString(), true);
1145+
1146+
String registerFunction = switch (operator.metadata.viewKind) {
1147+
case MATERIALIZED -> "register_materialized_output_zset_persistent";
1148+
case LOCAL -> throw new InternalCompilerError("Sink operator for local view " + operator);
1149+
case STANDARD -> "register_output_zset_persistent";
1150+
};
1151+
this.builder.append("catalog.")
1152+
.append(registerFunction)
1153+
.append("::<_, ");
1154+
operator.originalRowType.accept(this.innerVisitor);
1155+
this.builder.append(">(").increase().append("hash,")
1156+
.newline()
1157+
.append(this.getInputName(operator, 0))
1158+
.append(".clone()")
1159+
.append(",\n");
1160+
schemaJson.accept(this.innerVisitor);
1161+
this.builder.decrease().newline().append(");")
1162+
.newline();
1163+
}
11101164
}
11111165
if (this.options.ioOptions.sqlNames) {
11121166
this.builder.append("let ")
@@ -1117,6 +1171,7 @@ public VisitDecision preorder(DBSPSinkOperator operator) {
11171171
.newline();
11181172
}
11191173
} else {
1174+
// no handles
11201175
this.computeHash(operator);
11211176
this.builder.append("let ")
11221177
.append(this.handleName(operator))
@@ -1860,8 +1915,8 @@ public VisitDecision preorder(DBSPNowOperator operator) {
18601915
// Keeps track of tables that have to be materialized at a different point in the circuit.
18611916
final Bijection<DBSPSourceMultisetOperator, DBSPControlledKeyFilterOperator> materialization = new Bijection<>();
18621917

1918+
/** Discover input nodes that immediately feed the left input of a controlled_key_filter operator. */
18631919
void discoverLateMaterializations(DBSPCircuit circuit) {
1864-
// Discover input nodes that immediately feed the left input of a controlled_key_filter operator.
18651920
// This pattern is generated by inputs with LATENESS annotations.
18661921
// If the input table is materialized, the materialized stream has
18671922
// to be the output of the controlled_key_filter operator.
@@ -1878,11 +1933,71 @@ void discoverLateMaterializations(DBSPCircuit circuit) {
18781933
}
18791934
}
18801935

1936+
/** A view and all its indexes */
1937+
static class ViewAndIndexes {
1938+
@Nullable DBSPSinkOperator view;
1939+
List<DBSPSinkOperator> indexes = new ArrayList<>();
1940+
1941+
ViewAndIndexes() {
1942+
this.view = null;
1943+
}
1944+
1945+
void setView(DBSPSinkOperator view) {
1946+
Utilities.enforce(!view.isIndex());
1947+
Utilities.enforce(this.view == null);
1948+
this.view = view;
1949+
}
1950+
1951+
void addIndex(DBSPSinkOperator index) {
1952+
Utilities.enforce(index.isIndex());
1953+
this.indexes.add(index);
1954+
}
1955+
1956+
@Nullable
1957+
public DBSPSinkOperator firstIndex() {
1958+
if (this.indexes.isEmpty())
1959+
return null;
1960+
return this.indexes.get(0);
1961+
}
1962+
1963+
public boolean isEmpty() {
1964+
return this.indexes.isEmpty();
1965+
}
1966+
1967+
public DBSPSinkOperator getView() {
1968+
return Objects.requireNonNull(this.view);
1969+
}
1970+
}
1971+
1972+
// Key is a view name
1973+
final Map<String, ViewAndIndexes> indexes = new HashMap<>();
1974+
1975+
/** Group all indexes of a view in the 'indexes' map. */
1976+
public void discoverIndexes(DBSPCircuit circuit) {
1977+
for (DBSPOperator op: circuit.getAllOperators()) {
1978+
if (op.is(DBSPSinkOperator.class)) {
1979+
DBSPSinkOperator sink = op.to(DBSPSinkOperator.class);
1980+
if (sink.isIndex()) {
1981+
String view = sink.query;
1982+
if (!this.indexes.containsKey(view))
1983+
this.indexes.put(view, new ViewAndIndexes());
1984+
this.indexes.get(view).addIndex(sink);
1985+
} else {
1986+
String view = sink.viewName.name();
1987+
if (!this.indexes.containsKey(view))
1988+
this.indexes.put(view, new ViewAndIndexes());
1989+
this.indexes.get(view).setView(sink);
1990+
}
1991+
}
1992+
}
1993+
}
1994+
18811995
@Override
18821996
public Token startVisit(IDBSPOuterNode node) {
18831997
Token result = super.startVisit(node);
18841998
DBSPCircuit circuit = node.to(DBSPCircuit.class);
18851999
this.discoverLateMaterializations(circuit);
2000+
this.discoverIndexes(circuit);
18862001
return result;
18872002
}
18882003
}

‎sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/SingleOperatorWriter.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,8 @@ public void write(DBSPCompiler compiler) {
9494
ToRustVisitor visitor = new ToRustVisitor(
9595
compiler, this.builder(), this.circuit.metadata, new ProjectDeclarations(), this.materializations)
9696
.withPreferHash(true);
97+
visitor.discoverIndexes(circuit);
98+
9799
final String name = this.operator.getNodeName(true);
98100
this.builder().newline();
99101
for (String dep: this.dependencies)

‎sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/util/Utilities.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@
6666
import java.util.Objects;
6767
import java.util.Set;
6868
import java.util.function.Supplier;
69+
import java.util.regex.Pattern;
6970
import java.util.zip.ZipEntry;
7071
import java.util.zip.ZipInputStream;
7172

@@ -650,4 +651,9 @@ public static List<String> expandBraces(String pattern) {
650651
public static boolean inCI() {
651652
return System.getenv("CI") != null;
652653
}
654+
655+
/** Return the number of times 'pattern' occurs in 'string' (non-overlapping matches) */
656+
public static int countMatches(String string, String pattern) {
657+
return string.split(Pattern.quote(pattern), -1).length - 1;
658+
}
653659
}

0 commit comments

Comments
 (0)