Replace ExplodeExpression with a table function over an ExplodeValue

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/metadata/UnnestedRecordType.java

@@ -39,9 +39,10 @@
 import com.apple.foundationdb.record.query.plan.cascades.GraphExpansion;
 import com.apple.foundationdb.record.query.plan.cascades.NullableArrayTypeUtils;
 import com.apple.foundationdb.record.query.plan.cascades.Quantifier;
 import com.apple.foundationdb.record.query.plan.cascades.Reference;
-import com.apple.foundationdb.record.query.plan.cascades.expressions.ExplodeExpression;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.TableFunctionExpression;
 import com.apple.foundationdb.record.query.plan.cascades.typing.Type;
+import com.apple.foundationdb.record.query.plan.cascades.values.ExplodeValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.FieldValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.PromoteValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.RecordConstructorValue;
@@ -386,8 +387,8 @@
                         constituentQuantifier(ownerElementValue, constituent.getNestingExpression());
                 builder.addQuantifier(constituentQuantifier);
                 final Value flowedValue = constituentQuantifier.getFlowedObjectValue();
-                elementValue = FieldValue.ofOrdinalNumber(flowedValue, 0);
-                final Value positionValue = PromoteValue.inject(FieldValue.ofOrdinalNumber(flowedValue, 1),
+                elementValue = ExplodeValue.elementValue(flowedValue);
+                final Value positionValue = PromoteValue.inject(ExplodeValue.ordinalValue(flowedValue),
                         Type.primitiveType(Type.TypeCode.LONG, false));
                 positionColumns.add(Column.of(Optional.of(constituent.getName()), positionValue));
             }
@@ -402,10 +403,9 @@
     /**
      * Builds the quantifier standing for one constituent: a select over an explode of the constituent's array.
      *
-     * <p>The explode is created {@code WITH ORDINALITY}, so it flows an anonymous {@code (element, ordinal)} struct.
-     * The ordinal is needed to reconstruct the {@code __positions} field of a synthetic record, without which the
-     * synthetic primary key cannot be expressed. The struct is asked for as a record constructor, so that the
-     * element and the ordinal are sub-values a query's values can be matched against.
+     * <p>The explode is an {@link ExplodeValue}, so it flows an anonymous {@code (element, ordinal)} struct. The ordinal
+     * is needed to reconstruct the {@code __positions} field of a synthetic record, so the explode preserves the order
+     * of the array.
      *
      * @param ownerElementValue the record the array hangs off
      * @param nestingExpression the constituent's nesting expression
@@ -415,8 +415,8 @@
     private static Quantifier.ForEach constituentQuantifier(@Nonnull final Value ownerElementValue,
                                                             @Nonnull final KeyExpression nestingExpression) {
         final Quantifier.ForEach explodeQuantifier =
-                Quantifier.forEach(Reference.initialOf(new ExplodeExpression(
-                        FieldValue.ofFieldNames(ownerElementValue, arrayFieldPath(nestingExpression)), true, true)));
+                Quantifier.forEach(Reference.initialOf(new TableFunctionExpression(new ExplodeValue(
+                        FieldValue.ofFieldNames(ownerElementValue, arrayFieldPath(nestingExpression)), true))));
         return Quantifier.forEach(Reference.initialOf(GraphExpansion.ofQuantifier(explodeQuantifier)
                 .seal()
                 .buildSimpleSelectOverQuantifier(explodeQuantifier)));

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/metadata/expressions/FieldKeyExpression.java

@@ -31,8 +31,10 @@
 import com.apple.foundationdb.record.query.expressions.Query;
 import com.apple.foundationdb.record.query.plan.cascades.Reference;
 import com.apple.foundationdb.record.query.plan.cascades.KeyExpressionVisitor;
 import com.apple.foundationdb.record.query.plan.cascades.Quantifier;
-import com.apple.foundationdb.record.query.plan.cascades.expressions.ExplodeExpression;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.TableFunctionExpression;
+import com.apple.foundationdb.record.query.plan.cascades.values.ExplodeValue;
+import com.apple.foundationdb.record.query.plan.cascades.values.FieldValue;
 import com.apple.foundationdb.record.util.ProtoUtils;
 import com.google.common.collect.ImmutableList;
 import com.google.protobuf.Descriptors;
@@ -280,13 +282,11 @@
                 .addAll(fieldNamePrefix)
                 .add(ProtoUtils.toUserIdentifier(fieldName))
                 .build();
-        switch (fanType) {
-            case FanOut:
-                return Quantifier.forEach(Reference.initialOf(
-                        ExplodeExpression.explodeField(baseQuantifier, fieldNames)));
-            default:
-                throw new RecordCoreException("unrecognized fan type");
+        if (fanType == FanType.FanOut) {
+            return Quantifier.forEach(Reference.initialOf(new TableFunctionExpression(
+                    new ExplodeValue(FieldValue.ofFieldNames(baseQuantifier.getFlowedObjectValue(), fieldNames)))));
         }
+        throw new RecordCoreException("unrecognized fan type");
     }
 
     @Nonnull

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/expressions/OneOfThemWithComparison.java

@@ -28,8 +28,11 @@
 import com.apple.foundationdb.record.query.plan.cascades.Reference;
 import com.apple.foundationdb.record.query.plan.cascades.GraphExpansion;
 import com.apple.foundationdb.record.query.plan.cascades.Quantifier;
-import com.apple.foundationdb.record.query.plan.cascades.expressions.ExplodeExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.SelectExpression;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.TableFunctionExpression;
+import com.apple.foundationdb.record.query.plan.cascades.values.ExplodeValue;
+import com.apple.foundationdb.record.query.plan.cascades.values.FieldValue;
+import com.apple.foundationdb.record.query.plan.cascades.values.Value;
 import com.google.common.collect.ImmutableList;
 import com.google.protobuf.Descriptors;
 import com.google.protobuf.Message;
@@ -107,13 +110,15 @@
                 .addAll(fieldNamePrefix)
                 .add(getFieldName())
                 .build();
-        final Quantifier.ForEach childBase = Quantifier.forEach(Reference.initialOf(ExplodeExpression.explodeField(baseQuantifier, fieldNames)));
+        final Quantifier.ForEach childBase = Quantifier.forEach(Reference.initialOf(new TableFunctionExpression(
+                new ExplodeValue(FieldValue.ofFieldNames(baseQuantifier.getFlowedObjectValue(), fieldNames)))));
+        final Value elementValue = ExplodeValue.elementValue(childBase.getFlowedObjectValue());
         final SelectExpression selectExpression =
                 GraphExpansion.builder()
                         .addQuantifier(childBase)
-                        .addPredicate(childBase.getFlowedObjectValue().withComparison(comparison))
+                        .addPredicate(elementValue.withComparison(comparison))
                         .build()
-                        .buildSimpleSelectOverQuantifier(childBase);
+                        .buildSelectWithResultValue(elementValue);
         final Quantifier.Existential childQuantifier = Quantifier.existential(Reference.initialOf(selectExpression));
 
         // create a query component that creates a path to this prefix and then applies this to it

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/expressions/OneOfThemWithComponent.java

@@ -28,15 +28,16 @@
 import com.apple.foundationdb.record.query.plan.cascades.Reference;
 import com.apple.foundationdb.record.query.plan.cascades.GraphExpansion;
 import com.apple.foundationdb.record.query.plan.cascades.Quantifier;
-import com.apple.foundationdb.record.query.plan.cascades.expressions.ExplodeExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.SelectExpression;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.TableFunctionExpression;
+import com.apple.foundationdb.record.query.plan.cascades.values.ExplodeValue;
+import com.apple.foundationdb.record.query.plan.cascades.values.FieldValue;
 import com.google.common.collect.ImmutableList;
 import com.google.protobuf.Descriptors;
 import com.google.protobuf.Message;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
-import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.function.Supplier;
@@ -108,11 +109,13 @@
                 .addAll(fieldNamePrefix)
                 .add(getFieldName())
                 .build();
-        final Quantifier.ForEach childBase = Quantifier.forEach(Reference.initialOf(ExplodeExpression.explodeField(baseQuantifier, fieldNames)));
-        final GraphExpansion graphExpansion = getChild().expand(childBase, outerQuantifierSupplier, Collections.emptyList());
+        final Quantifier.ForEach childBase = Quantifier.forEach(Reference.initialOf(new TableFunctionExpression(
+                new ExplodeValue(FieldValue.ofFieldNames(baseQuantifier.getFlowedObjectValue(), fieldNames)))));
+        final GraphExpansion graphExpansion =
+                getChild().expand(childBase, outerQuantifierSupplier, ImmutableList.of(ExplodeValue.ELEMENT_FIELD_NAME));
         final SelectExpression selectExpression =
                 GraphExpansion.ofOthers(GraphExpansion.builder().addQuantifier(childBase).build(), graphExpansion)
-                        .buildSimpleSelectOverQuantifier(childBase);
+                        .buildSelectWithResultValue(ExplodeValue.elementValue(childBase.getFlowedObjectValue()));
 
         Quantifier.Existential childQuantifier = Quantifier.existential(Reference.initialOf(selectExpression));
 

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/Compensation.java

@@ -1037,21 +1037,26 @@
         }
 
         @Nonnull
         private CorrelationIdentifier getMatchedForEachAlias() {
+            return Iterables.getOnlyElement(getMatchedForEachAliases());
+        }
+
+        /**
+         * {@return the aliases of the matched for-each quantifiers}
+         */
+        @Nonnull
+        private Set<CorrelationIdentifier> getMatchedForEachAliases() {
             final var matchedQuantifierMap =
                     Quantifiers.aliasToQuantifierMap(matchedQuantifiers);
 
             final var matchedAliases =
                     matchedQuantifierMap.keySet();
             Verify.verify(compensatedAliases.equals(matchedAliases));
 
-            final var matchedForEachQuantifierAliases =
-                    matchedAliases
-                            .stream()
-                            .filter(alias -> matchedQuantifierMap.get(alias) instanceof Quantifier.ForEach)
-                            .collect(ImmutableSet.toImmutableSet());
-
-            return Iterables.getOnlyElement(matchedForEachQuantifierAliases);
+            return matchedAliases
+                    .stream()
+                    .filter(alias -> matchedQuantifierMap.get(alias) instanceof Quantifier.ForEach)
+                    .collect(ImmutableSet.toImmutableSet());
         }
 
         @Nonnull
@@ -1062,7 +1067,9 @@
             Verify.verify(!isImpossible());
             Verify.verify(resultCompensationFunction.isNeeded());
 
-            final var matchedForEachAlias = getMatchedForEachAlias();
+            // The compensated result value is expressed over the top of the match, so the alias picked here only names
+            // the quantifier it is re-targeted onto -- any matched for-each alias will do, and a join match has several.
+            final var matchedForEachAlias = Iterables.get(getMatchedForEachAliases(), 0);
 
             final var resultValue =
                     resultCompensationFunction.applyCompensationForResult(

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/KeyExpressionExpansionVisitor.java

@@ -36,8 +36,10 @@
 import com.apple.foundationdb.record.query.plan.cascades.KeyExpressionExpansionVisitor.VisitorState;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.SelectExpression;
 import com.apple.foundationdb.record.query.plan.cascades.predicates.Placeholder;
 import com.apple.foundationdb.record.query.plan.cascades.predicates.PredicateWithValueAndRanges;
+import com.apple.foundationdb.record.query.plan.cascades.typing.Type;
 import com.apple.foundationdb.record.query.plan.cascades.values.EmptyValue;
+import com.apple.foundationdb.record.query.plan.cascades.values.ExplodeValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.FieldValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.Value;
 import com.apple.foundationdb.record.util.ProtoUtils;
@@ -130,59 +132,66 @@
     @Nonnull
     @Override
     public GraphExpansion visitExpression(@Nonnull FieldKeyExpression fieldKeyExpression) {
         final String fieldName = ProtoUtils.toUserIdentifier(fieldKeyExpression.getFieldName());
         final KeyExpression.FanType fanType = fieldKeyExpression.getFanType();
         final VisitorState state = getCurrentState();
         final List<String> fieldNamePrefix = state.getFieldNamePrefix();
         final Quantifier.ForEach baseQuantifier = state.getBaseQuantifier();
         final List<String> fieldNames = ImmutableList.<String>builder()
                 .addAll(fieldNamePrefix)
                 .add(fieldName)
                 .build();
         final Value value;
         final Column<?> column;
         switch (fanType) {
             case FanOut:
                 // explode this field and prefixes of this field
                 final Quantifier.ForEach childBase = fieldKeyExpression.explodeField(baseQuantifier, fieldNamePrefix);
-                value = state.registerValue(childBase.getFlowedObjectValue());
-                column = Column.unnamedOf(value);
+                value = state.registerValue(ExplodeValue.elementValue(childBase.getFlowedObjectValue()));
+                //
+                // The select over the explode flows the explode's whole (element, ordinal) struct rather than just the
+                // element, so that a query's select over the explode matches it whether it flows the element or the
+                // struct. The element is then accessed on the struct to form the column of this expansion.
+                //
+                column = Column.unnamedOf(childBase.getFlowedObjectValue());
                 final GraphExpansion childExpansion;
                 if (state.isKey() && !state.isInternalExpansion()) {
                     childExpansion = GraphExpansion.ofResultColumnAndPlaceholder(column, value.asPlaceholder(newParameterAlias()));
                 } else {
                     childExpansion = GraphExpansion.ofResultColumn(column);
                 }
                 final SelectExpression selectExpression =
                         childExpansion
                                 .withBase(childBase)
                                 .buildSelect();
                 final Quantifier childQuantifier = Quantifier.forEach(Reference.initialOf(selectExpression));
                 final GraphExpansion.Sealed sealedChildExpansion = childExpansion.seal();
                 return sealedChildExpansion
                         .builderWithInheritedPlaceholders()
-                        .pullUpQuantifier(childQuantifier)
+                        .addQuantifier(childQuantifier)
+                        .addResultColumn(Column.unnamedOf(ExplodeValue.elementValue(
+                                FieldValue.ofOrdinalNumber(childQuantifier.getFlowedObjectValue(), 0))))
                         .build();
             case Concatenate:
             case None:
                 // Note: `Concatenate` and `None` can use the graph expansion logic. Both just access the field directly
                 // via `FieldValue.ofFieldNames()`.
                 value = state.registerValue(FieldValue.ofFieldNames(baseQuantifier.getFlowedObjectValue(), fieldNames));
                 if (state.isSelectStar()) {
                     if (state.isKey() && !state.isInternalExpansion()) {
                         return GraphExpansion.ofPlaceholder(value.asPlaceholder(newParameterAlias()));
                     }
                     return GraphExpansion.empty();
                 } else {
                     column = Column.unnamedOf(value);
                     if (state.isKey() && !state.isInternalExpansion()) {
                         return GraphExpansion.ofResultColumnAndPlaceholder(column, value.asPlaceholder(newParameterAlias()));
                     }
                     return GraphExpansion.ofResultColumn(column);
                 }
             default:
                 throw new UnsupportedOperationException();
         }
     }
 
     @Nonnull
@@ -311,12 +320,12 @@
                 final Quantifier.ForEach childBaseQuantifier = parent.explodeField(baseQuantifier, fieldNamePrefix);
                 // expand the children of the key expression and then unify them into an expansion of this expression
                 final GraphExpansion childExpansion =
-                        pop(child.expand(push(state.withBaseQuantifier(childBaseQuantifier).withFieldNamePrefix(ImmutableList.of()))));
+                        pop(child.expand(push(state.withBaseQuantifier(childBaseQuantifier)
+                                .withFieldNamePrefix(ImmutableList.of(ExplodeValue.ELEMENT_FIELD_NAME)))));
                 final GraphExpansion baseAndChildExpansion;
                 if (state.isSelectStar()) {
                     final GraphExpansion baseExpansion =
-                            GraphExpansion.builder()
-                                    .pullUpQuantifier(childBaseQuantifier)
+                            pullUpExplodedElement(GraphExpansion.builder(), childBaseQuantifier)
                                     .build();
                     baseAndChildExpansion = GraphExpansion.ofOthers(baseExpansion, childExpansion);
 
@@ -339,7 +348,7 @@
                             GraphExpansion.builder()
                                     .addAllPredicates(childExpansion.getPredicates())
                                     .addAllPlaceholders(childExpansion.getPlaceholders());
-                    baseAndChildExpansionBuilder.pullUpQuantifier(childBaseQuantifier);
+                    pullUpExplodedElement(baseAndChildExpansionBuilder, childBaseQuantifier);
                     childExpansion.getQuantifiers()
                             .forEach(baseAndChildExpansionBuilder::pullUpQuantifier);
                     baseAndChildExpansion = baseAndChildExpansionBuilder.build();
@@ -369,6 +378,34 @@
         }
     }
 
+    /**
+     * Adds a quantifier ranging over an explode to the builder and pulls up the columns of the exploded element followed
+     * by the ordinal, rather than the columns of the struct the explode flows. The struct holds the element in its first
+     * column, so pulling up the struct would yield a single opaque column for the element where a query's select over
+     * the same explode flows the element's fields, which then do not line up with each other. The ordinal is pulled up
+     * too, so that a query's select that flows the whole struct can still form it from the columns.
+     *
+     * @param builder the builder to add to
+     * @param explodeQuantifier a quantifier ranging over an {@link ExplodeValue}
+     * @return the {@code builder}
+     */
+    @Nonnull
+    private static GraphExpansion.Builder pullUpExplodedElement(@Nonnull final GraphExpansion.Builder builder,
+                                                                @Nonnull final Quantifier.ForEach explodeQuantifier) {
+        builder.addQuantifier(explodeQuantifier);
+        final Value elementValue = ExplodeValue.elementValue(explodeQuantifier.getFlowedObjectValue());
+        if (elementValue.getResultType() instanceof Type.Record) {
+            final var fields = Objects.requireNonNull(((Type.Record)elementValue.getResultType()).getFields());
+            for (int i = 0; i < fields.size(); i++) {
+                builder.addResultColumn(Column.of(fields.get(i), FieldValue.ofOrdinalNumberAndFuseIfPossible(elementValue, i)));
+            }
+        } else {
+            builder.addResultColumn(Column.unnamedOf(elementValue));
+        }
+        builder.addResultColumn(Column.unnamedOf(ExplodeValue.ordinalValue(explodeQuantifier.getFlowedObjectValue())));
+        return builder;
+    }
+
     @Nonnull
     private static ImmutableList<Placeholder> pullUpPlaceholders(@Nonnull final GraphExpansion childExpansion,
                                                                  @Nonnull final Value childResultValue,

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/PlanningRuleSet.java

@@ -28,7 +28,6 @@
 import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementDeleteRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementDistinctRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementDistinctUnionRule;
-import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementExplodeRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementFilterRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementInJoinRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementInUnionRule;
@@ -164,13 +163,12 @@
             new ImplementInJoinRule(),
             new ImplementInUnionRule(),
             new ImplementSimpleSelectRule(),
-            new ImplementExplodeRule(),
             new ImplementNestedLoopJoinRule(),
             new ImplementStreamingAggregationRule(),
             new ImplementDeleteRule(),
             new ImplementInsertRule(),
             new ImplementTempTableInsertRule(),
             new ImplementUpdateRule(),
             new ImplementRecursiveLevelUnionRule(),
             new ImplementRecursiveDfsJoinRule(),
             new ImplementTableFunctionRule()