Skip to content

[SQL] Retry compiling a query with CTEs by converting CTEs to local views #6836

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

Merged
mihaibudiu merged 1 commit into from
Aug 10, 2026
Merged

[SQL] Retry compiling a query with CTEs by converting CTEs to local views #6836

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
8 changes: 7 additions & 1 deletion ...sp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/CompilerOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,11 @@ public static class Language implements IDiff<Language>, IValidate {
@Parameter(names = "--unaryPlusNoop",
description = "Compile unary plus into a no-operation; similar to sqlite")
public boolean unaryPlusNoop = false;
/** Rewrite CTE queries to use local views instead.
* Hidden: intended for testing and debugging. */
@Parameter(names = "--cteViews", hidden = true,
description = "Convert each top-level common table expression (WITH) in a view into a LOCAL VIEW")
public boolean cteViews = false;

public boolean same(Language language) {
// Only compare fields that matter.
Expand All @@ -84,7 +89,8 @@ public boolean same(Language language) {
@Override
public String toString() {
return "Language{" +
"\n\tgenerateInputForEveryTable=" + this.generateInputForEveryTable +
"\n\tcteViews=" + this.cteViews +
",\n\tgenerateInputForEveryTable=" + this.generateInputForEveryTable +
",\n\tignoreOrderBy=" + this.ignoreOrderBy +
",\n\tincrementalize=" + this.incrementalize +
",\n\tlenient=" + this.lenient +
Expand Down
80 changes: 60 additions & 20 deletions ...-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/DBSPCompiler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@
import org.dbsp.sqlCompiler.compiler.errors.SourcePositionRange;
import org.dbsp.sqlCompiler.compiler.errors.UnsupportedException;
import org.dbsp.sqlCompiler.compiler.frontend.TypeCompiler;
import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.CteToLocalViews;
import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.ForeignKey;
import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.ParsedStatement;
import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.ProgramIdentifier;
Expand Down Expand Up @@ -587,6 +588,42 @@ void emitSql(List<ParsedStatement> statements) {
outputStream.close();
}

/** Compile a CREATE VIEW statement. The view's top-level common table
* expressions may be converted to LOCAL VIEWs. The result contains
* one statement per converted CTE, followed by the view itself. */
List<RelStatement> compileCreateView(ParsedStatement node, Map<ProgramIdentifier, SqlLateness> lateness) {
@Nullable List<ParsedStatement> parts = null;
if (this.options.languageOptions.cteViews)
parts = this.sqlToRelCompiler.hoistCtes(node);
if (parts == null) {
try {
RelStatement fe = this.sqlToRelCompiler.compileCreateView(
node, lateness, this.sources, /* retry */ true);
return fe == null ? Linq.list() : Linq.list(fe);
} catch (CteToLocalViews.Retry retry) {
Logger.INSTANCE.belowLevel(this, 1)
.append("Could not decorrelate query; retrying with CTEs as local views")
.newline();
parts = this.sqlToRelCompiler.hoistCtes(node);
}
if (parts == null) {
// The rewrite turned out not to apply; compile again without
// retry, so the real error surfaces.
RelStatement fe = this.sqlToRelCompiler.compileCreateView(
node, lateness, this.sources, /* retry */ false);
return fe == null ? Linq.list() : Linq.list(fe);
}
}
List<RelStatement> result = new ArrayList<>();
ParsedStatement view = Utilities.last(parts);
for (ParsedStatement part: parts) {
// Lateness declarations apply to the original view only
Map<ProgramIdentifier, SqlLateness> lat = part == view ? lateness : new HashMap<>();
result.addAll(this.compileCreateView(part, lat));
}
return result;
}

@Nullable DBSPCircuit runAllCompilerStages() {
List<ParsedStatement> parsed = this.runParser();
if (this.hasErrors())
Expand Down Expand Up @@ -694,32 +731,35 @@ void emitSql(List<ParsedStatement> statements) {
if (node.statement() instanceof SqlLateness)
continue;

RelStatement fe;
List<RelStatement> compiled;
if (node.statement() instanceof SqlCreateView cv) {
ProgramIdentifier viewName = ProgramIdentifier.fromSqlId(cv.name);
Map<ProgramIdentifier, SqlLateness> late = this.viewLateness.getOrDefault(viewName, new HashMap<>());
fe = this.sqlToRelCompiler.compileCreateView(node, late, this.sources);
Map<ProgramIdentifier, SqlLateness> lateness = this.viewLateness.getOrDefault(viewName, new HashMap<>());
compiled = this.compileCreateView(node, lateness);
} else {
fe = this.sqlToRelCompiler.compile(node, this.sources);
RelStatement single = this.sqlToRelCompiler.compile(node, this.sources);
compiled = single == null
// error during compilation
? Linq.list()
: Linq.list(single);
}
if (fe == null)
// error during compilation
continue;

if (fe.is(CreateViewStatement.class)) {
CreateViewStatement cv = fe.to(CreateViewStatement.class);
Utilities.putNew(this.views, cv.getName(), cv);
} else if (fe.is(CreateTableStatement.class)) {
CreateTableStatement ct = fe.to(CreateTableStatement.class);
foreignKeys.addAll(ct.foreignKeys);
} else if (fe.is(CreateIndexStatement.class)) {
CreateIndexStatement ct = fe.to(CreateIndexStatement.class);
boolean success = this.validateCreateIndex(ct);
if (!success)
return null;
Utilities.putNew(this.indexes, ct.getName(), ct);
for (RelStatement fe: compiled) {
if (fe.is(CreateViewStatement.class)) {
CreateViewStatement cv = fe.to(CreateViewStatement.class);
Utilities.putNew(this.views, cv.getName(), cv);
} else if (fe.is(CreateTableStatement.class)) {
CreateTableStatement ct = fe.to(CreateTableStatement.class);
foreignKeys.addAll(ct.foreignKeys);
} else if (fe.is(CreateIndexStatement.class)) {
CreateIndexStatement ct = fe.to(CreateIndexStatement.class);
boolean success = this.validateCreateIndex(ct);
if (!success)
return null;
Utilities.putNew(this.indexes, ct.getName(), ct);
}
this.relToDBSPCompiler.compile(fe);
}
this.relToDBSPCompiler.compile(fe);
}
this.setErrorContext(SourcePositionRange.INVALID);

Expand Down
53 changes: 44 additions & 9 deletions ...-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -372,6 +372,42 @@ UnimplementedException decorrelateError(CalciteObject node) {
"It looks like the compiler could not decorrelate this query.", 2555, node);
}

/** True if {@link #visitCorrelate} can implement this correlate
* (an INNER unnest-shaped correlate compiled into a flat_map). */
public static boolean isImplementableCorrelate(LogicalCorrelate correlate) {
if (correlate.getJoinType() != JoinRelType.INNER)
return false;
RelNode right = correlate.getRight();
if (right instanceof Project project)
right = project.getInput();
else if (right instanceof Filter filter)
right = filter.getInput();
return right instanceof Uncollect uncollect
&& uncollect.getInput() instanceof LogicalProject uncollectInput
&& uncollectInput.getProjects().size() == 1;
}

/** True if the plan contains a correlate that the compiler cannot
* implement; such a correlate means that the decorrelator has failed. */
public static boolean hasUnimplementableCorrelate(RelNode plan) {
class Finder extends RelVisitor {
boolean found = false;

@Override
public void visit(RelNode node, int ordinal, @org.checkerframework.checker.nullness.qual.Nullable RelNode parent) {
if (node instanceof LogicalCorrelate correlate
&& !isImplementableCorrelate(correlate)) {
this.found = true;
return;
}
super.visit(node, ordinal, parent);
}
}
Finder finder = new Finder();
finder.go(plan);
return finder.found;
}

void visitCorrelate(LogicalCorrelate correlate) {
/*
We decorrelate queries using Calcite's optimizer, which doesn't always work.
Expand Down Expand Up @@ -414,12 +450,16 @@ void visitCorrelate(LogicalCorrelate correlate) {
DBSPTypeTuple type = this.convertType(
node.getPositionRange(), correlate.getRowType(), false).to(DBSPTypeTuple.class);

if (correlate.getJoinType() != JoinRelType.INNER)
throw new UnimplementedException("LEFT JOIN UNNEST");
if (!isImplementableCorrelate(correlate)) {
if (correlate.getJoinType() != JoinRelType.INNER)
throw new UnimplementedException("LEFT JOIN UNNEST");
throw this.decorrelateError(node);
}
this.visit(correlate.getLeft(), 0, correlate);
DBSPSimpleOperator left = this.getInputAs(correlate.getLeft(), true);
DBSPTypeTuple leftElementType = left.getOutputZSetElementType().to(DBSPTypeTuple.class);

// The casts below are safe: isImplementableCorrelate checked the shape.
RelNode correlateRight = correlate.getRight();
Project rightProject = null;
Filter rightFilter = null;
Expand All @@ -430,13 +470,8 @@ void visitCorrelate(LogicalCorrelate correlate) {
rightFilter = (Filter) correlateRight;
correlateRight = rightFilter.getInput();
}
if (!(correlateRight instanceof Uncollect uncollect))
throw this.decorrelateError(node);
RelNode uncollectInput = uncollect.getInput();
if (!(uncollectInput instanceof LogicalProject project))
throw this.decorrelateError(node);
if (project.getProjects().size() != 1)
throw this.decorrelateError(node);
Uncollect uncollect = (Uncollect) correlateRight;
LogicalProject project = (LogicalProject) uncollect.getInput();
RexNode projection = project.getProjects().get(0);
DBSPVariablePath dataVar = new DBSPVariablePath(leftElementType.ref());
ExpressionCompiler eComp = new ExpressionCompiler(correlate, dataVar, this.compiler);
Expand Down
Loading