Support recursive CTEs

[kasiafi]

Adds support for recursive common table expression (WITH RECURSIVE).
Single-element recursive cycles are allowed.

Issue: #1122

[martint]

Some initial comments. Still reviewing.

[martint]

Improvement for the future: use the corresponding SELECT expression as the "node" in the exception so that error highlighting points at the right place. It won't be trivial to do, so let's defer it (add a TODO)

[kasiafi]

The number of the first mismatching column is included in the error message. It is not always possible to highlight the exact select expression, because it might be e.g. a "row all fields" expression. I'm not sure what I should write in the TODO.

[martint]

The many levels of nesting and this continue here make it hard to follow the flow of the code. One possible way of reorganizing this is to move this whole section to a new method and have it return "true" if it was able to handle the recursive query. Then, the code below would do:

if (!handled) {
   // 2. 
   ...
   // process plain query
   ...
}

Let me play with the code a bit to see if I can come up with something better.

[martint]

Also, instead of splitting the method into two sections based on with.isRecursive(), it seems we should be able to structure in a similar manner to this and avoid a bit of duplication:

With with = node.getWith().get();
Scope.Builder withScopeBuilder = scopeBuilder(scope);

for (WithQuery withQuery : with.getQueries()) {
    String name = withQuery.getName().getValue().toLowerCase(ENGLISH);
    if (withScopeBuilder.containsNamedQuery(name)) {
        throw semanticException(DUPLICATE_NAMED_QUERY, withQuery, "WITH query name '%s' specified more than once", name);
    }

    boolean isRecursive = false;
    if (with.isRecursive()) {
        isRecursive = tryProcessRecursiveQuery(...);
    }

    if (!isRecursive) {
        // check if all or none of the columns are explicitly alias
        Query query = withQuery.getQuery();
        if (withQuery.getColumnNames().isPresent()) {
            validateColumnAliases(withQuery.getColumnNames().get(), analysis.getOutputDescriptor(query).getVisibleFieldCount());
        }

        process(query, withScopeBuilder.build());
        withScopeBuilder.withNamedQuery(name, withQuery);
    }
}

Scope withScope = withScopeBuilder.build();
analysis.setScope(with, withScope);
return withScope;

Possibly, even this for the inner condition, but I'm not a big fan of the side-effect in the if condition:

...
if (!with.isRecursive() || !tryProcessRecursiveQuery(...)) {
   ...
}
...

[martint]

Looks good. Can you rebased and squash the commits?

[martint]

Do we need this anymore? With the field mappings as a reference, it doesn't matter how many fields the underlying plan node produces.

[kasiafi]

It is necessary to prune anchor outputs. If we leave those extra symbols in the anchor plan, they will "leak" to the recursion step plan. Later, when anchor plan is replaced with step plan, they will become orphaned.

[martint]

Using inheritance to share code is generally not a good approach. It introduces issues of poor encapsulation, unclear thread-safety semantics (for code that can be accessed concurrently), etc. It's better to do it via delegation. In this case, you could make SymbolMapper take an object during construction that is responsible for doing the primordial mapping. You'd customize that for each of the two use cases we need here.

[kasiafi]

Updated and squashed.