diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/DerivationsProperty.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/DerivationsProperty.java index b3ea80006b..ea94ca7e86 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/DerivationsProperty.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/DerivationsProperty.java @@ -39,6 +39,7 @@ import com.apple.foundationdb.record.query.plan.cascades.predicates.PredicateWithValue; import com.apple.foundationdb.record.query.plan.cascades.predicates.QueryPredicate; 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.FirstOrDefaultStreamingValue; import com.apple.foundationdb.record.query.plan.cascades.values.FirstOrDefaultValue; import com.apple.foundationdb.record.query.plan.cascades.values.QueriedValue; @@ -394,6 +395,16 @@ public Derivations visitInsertPlan(@Nonnull final RecordQueryInsertPlan insertPl @Override public Derivations visitTableFunctionPlan(@Nonnull final RecordQueryTableFunctionPlan tableFunctionPlan) { final var streamingValue = tableFunctionPlan.getValue(); + if (streamingValue instanceof final ExplodeValue explodeValue) { + // Use the first element of the collection as a representative for "some" element, and wrap it in the + // struct the explode flows, so that accesses of the element are written against the struct. + final var elementType = Objects.requireNonNull(((Type.Array)explodeValue.getCollectionValue().getResultType()).getElementType()); + final var columns = ImmutableList.>builder(); + columns.add(Column.unnamedOf(new FirstOrDefaultValue(explodeValue.getCollectionValue(), new ThrowsValue(elementType)))); + columns.add(Column.unnamedOf(new ThrowsValue(Type.primitiveType(Type.TypeCode.INT, false)))); + final var representatives = ImmutableList.of(RecordConstructorValue.ofColumns(columns.build(), false)); + return new Derivations(representatives, representatives); + } final var elementType = streamingValue.getResultType(); final var values = ImmutableList.of(new FirstOrDefaultStreamingValue(streamingValue, new ThrowsValue(elementType))); return new Derivations(values, values); diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/values/ExplodeValue.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/values/ExplodeValue.java new file mode 100644 index 0000000000..8ff2b7a140 --- /dev/null +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/values/ExplodeValue.java @@ -0,0 +1,528 @@ +/* + * ExplodeValue.java + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2015-2026 Apple Inc. and the FoundationDB project authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.apple.foundationdb.record.query.plan.cascades.values; + +import com.apple.foundationdb.annotation.API; +import com.apple.foundationdb.annotation.SpotBugsSuppressWarnings; +import com.apple.foundationdb.record.Bindings; +import com.apple.foundationdb.record.EvaluationContext; +import com.apple.foundationdb.record.ExecuteProperties; +import com.apple.foundationdb.record.ObjectPlanHash; +import com.apple.foundationdb.record.PlanDeserializer; +import com.apple.foundationdb.record.PlanHashable; +import com.apple.foundationdb.record.PlanSerializationContext; +import com.apple.foundationdb.record.RecordCoreException; +import com.apple.foundationdb.record.RecordCursor; +import com.apple.foundationdb.record.RecordCursorContinuation; +import com.apple.foundationdb.record.RecordCursorProto; +import com.apple.foundationdb.record.RecordCursorResult; +import com.apple.foundationdb.record.RecordCursorVisitor; +import com.apple.foundationdb.record.logging.LogMessageKeys; +import com.apple.foundationdb.record.planprotos.PExplodeValue; +import com.apple.foundationdb.record.planprotos.PValue; +import com.apple.foundationdb.record.provider.foundationdb.FDBRecordStoreBase; +import com.apple.foundationdb.record.query.plan.cascades.AliasMap; +import com.apple.foundationdb.record.query.plan.cascades.Column; +import com.apple.foundationdb.record.query.plan.cascades.ConstrainedBoolean; +import com.apple.foundationdb.record.query.plan.cascades.CorrelationIdentifier; +import com.apple.foundationdb.record.query.plan.cascades.properties.CardinalitiesProperty; +import com.apple.foundationdb.record.query.plan.cascades.typing.Type; +import com.apple.foundationdb.record.query.plan.explain.ExplainTokens; +import com.apple.foundationdb.record.query.plan.explain.ExplainTokensWithPrecedence; +import com.apple.foundationdb.record.query.plan.plans.QueryResult; +import com.apple.foundationdb.tuple.ByteArrayUtil2; +import com.google.auto.service.AutoService; +import com.google.common.base.Verify; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Iterables; +import com.google.protobuf.ByteString; +import com.google.protobuf.InvalidProtocolBufferException; +import com.google.protobuf.Message; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; +import java.util.function.BiFunction; +import java.util.function.Supplier; +import java.util.stream.IntStream; + +/** + * A {@link StreamingValue} that iterates over the elements of an array. For every element it produces a struct of two + * anonymous fields: the element, and the 0-based ordinal of the element within the array, akin to SQL's + * {@code UNNEST ... WITH ORDINALITY}. + * + *

The struct has the same shape whether or not the explode {@linkplain #preservesOrder() preserves the order} of + * its collection. An explode that preserves the order produces the elements in the order of the array, so each ordinal + * is the position of its element in the array, and the planner must not reorder the collection. Any other explode + * does not observe its ordinals, so its collection can be replaced by any permutation of it, for example by a sorted + * one, even though that renumbers the ordinals it produces. + * + *

An explode that is {@linkplain #isSorted() sorted} produces the elements of its collection in ascending order, or + * in descending order if it is also {@linkplain #isReverse() reverse}, and numbers the ordinals in that order. The + * order is a property of the explode rather than of its collection, so it does not have to be recovered from whatever + * computes the collection. Only an explode that does not preserve the order of its collection can be sorted. + */ +@API(API.Status.EXPERIMENTAL) +public class ExplodeValue extends AbstractValue implements StreamingValue, CreatesDynamicTypesValue { + private static final ObjectPlanHash BASE_HASH = new ObjectPlanHash("Explode-Value"); + + @Nonnull + public static final String ELEMENT_FIELD_NAME = "_0"; + + @Nonnull + private static final Type ORDINAL_TYPE = Type.primitiveType(Type.TypeCode.INT, false); + + @Nonnull + @SuppressWarnings("unchecked") + private static final Comparator ELEMENT_COMPARATOR = Comparator.comparing(Comparable.class::cast); + + @Nonnull + private final Value collectionValue; + + @Nonnull + private final Type elementType; + + /** + * Whether the planner must keep the collection in its order. If set, every ordinal this explode produces is the + * position of its element in the array, so the collection must not be replaced by a permutation of it, for example + * by a sorted one. If not set, the ordinals are not observed and the collection may be reordered. Either way the + * explode flows the same {@code (element, ordinal)} struct. + */ + private final boolean preserveOrder; + + /** + * Whether the explode produces the elements of its collection sorted rather than in the order of the array. + */ + private final boolean sorted; + + /** + * Whether a sorted explode sorts in descending rather than ascending order. + */ + private final boolean reverse; + + @Nonnull + private final CorrelationIdentifier elementAlias; + + @Nonnull + private final CorrelationIdentifier ordinalAlias; + + @Nonnull + private final RecordConstructorValue structValue; + + public ExplodeValue(@Nonnull final Value collectionValue) { + this(collectionValue, false); + } + + public ExplodeValue(@Nonnull final Value collectionValue, final boolean preserveOrder) { + this(collectionValue, preserveOrder, false, false); + } + + public ExplodeValue(@Nonnull final Value collectionValue, final boolean preserveOrder, final boolean sorted, + final boolean reverse) { + Verify.verify(collectionValue.getResultType().isArray()); + Verify.verify(!preserveOrder || !sorted, "cannot sort an explode that preserves its order"); + Verify.verify(sorted || !reverse, "cannot reverse an explode that is not sorted"); + this.collectionValue = collectionValue; + this.preserveOrder = preserveOrder; + this.sorted = sorted; + this.reverse = reverse; + this.elementType = Objects.requireNonNull(((Type.Array)collectionValue.getResultType()).getElementType()); + Verify.verify(!sorted || isSortable(elementType), "cannot sort elements of type %s", elementType); + this.elementAlias = CorrelationIdentifier.uniqueId(); + this.ordinalAlias = CorrelationIdentifier.uniqueId(); + final ImmutableList.Builder> columnsBuilder = ImmutableList.builder(); + columnsBuilder.add(Column.unnamedOf(QuantifiedObjectValue.of(elementAlias, elementType))); + columnsBuilder.add(Column.unnamedOf(QuantifiedObjectValue.of(ordinalAlias, ORDINAL_TYPE))); + this.structValue = RecordConstructorValue.ofColumns(columnsBuilder.build(), false); + Verify.verify(ELEMENT_FIELD_NAME.equals(structValue.getResultType().getField(0).getFieldName())); + } + + @Nonnull + public Value getCollectionValue() { + return collectionValue; + } + + public boolean preservesOrder() { + return preserveOrder; + } + + public boolean isSorted() { + return sorted; + } + + public boolean isReverse() { + return reverse; + } + + /** + * Whether elements of a type can be sorted, which needs their representation to be {@link Comparable}. + * + * @param elementType the type of the elements + * @return {@code true} if an explode of an array of such elements can be sorted + */ + public static boolean isSortable(@Nonnull final Type elementType) { + if (elementType.isEnum()) { + return true; + } + return elementType.isPrimitive() && + elementType.getTypeCode() != Type.TypeCode.BYTES && elementType.getTypeCode() != Type.TypeCode.VECTOR; + } + + /** + * Returns an explode of a different collection that preserves its order, and is sorted, if and only if this one + * does and is. + * + * @param newCollectionValue the new collection + * @return an explode of the new collection + */ + @Nonnull + public ExplodeValue withCollectionValue(@Nonnull final Value newCollectionValue) { + return new ExplodeValue(newCollectionValue, preserveOrder, sorted, reverse); + } + + /** + * Returns an explode of the same collection that produces its elements sorted. Only an explode that does not + * preserve the order of its collection can be sorted. + * + * @param reverse whether to sort in descending order + * @return a sorted explode of the same collection + */ + @Nonnull + public ExplodeValue sorted(final boolean reverse) { + Verify.verify(!preserveOrder, "cannot sort an explode that preserves its order"); + return new ExplodeValue(collectionValue, false, true, reverse); + } + + @Nonnull + @Override + public Type.Record getResultType() { + return structValue.getResultType(); + } + + @Nonnull + @Override + public RecordCursor evalAsStream(@Nonnull final FDBRecordStoreBase store, + @Nonnull final EvaluationContext context, + @Nullable final byte[] continuation, + @Nonnull final ExecuteProperties executeProperties) { + return new Cursor(store.getExecutor(), elementsOf(store, context), + (elements, ordinal) -> structAt(context, elements, ordinal), continuation) + .skipThenLimit(executeProperties.getSkip(), executeProperties.getReturnedRowLimit()); + } + + @Nonnull + public List evalAsList(@Nullable final FDBRecordStoreBase store, @Nonnull final EvaluationContext context) { + final List elements = elementsOf(store, context); + return IntStream.range(0, elements.size()) + .mapToObj(ordinal -> structAt(context, elements, ordinal)) + .collect(ImmutableList.toImmutableList()); + } + + /** + * Evaluates the collection, and sorts it if the explode is sorted. A {@code null} collection has no elements. + */ + @Nonnull + private List elementsOf(@Nullable final FDBRecordStoreBase store, @Nonnull final EvaluationContext context) { + final Object result = collectionValue.eval(store, context); + if (result == null) { + return List.of(); + } + final List elements = (List)result; + if (!sorted) { + return elements; + } + final List sortedElements = new ArrayList<>(elements); + sortedElements.sort(reverse ? ELEMENT_COMPARATOR.reversed() : ELEMENT_COMPARATOR); + return sortedElements; + } + + @Nonnull + private Object structAt(@Nonnull final EvaluationContext context, @Nonnull final List elements, final int ordinal) { + return structOf(context, Verify.verifyNotNull(elements.get(ordinal), "array elements must be non-null"), ordinal); + } + + @Nonnull + private Object structOf(@Nonnull final EvaluationContext context, @Nonnull final Object element, final int ordinal) { + final EvaluationContext elementContext = context + .withBinding(Bindings.Internal.CORRELATION, elementAlias, QueryResult.ofComputed(element)) + .withBinding(Bindings.Internal.CORRELATION, ordinalAlias, QueryResult.ofComputed(ordinal)); + return Objects.requireNonNull(structValue.eval(null, elementContext)); + } + + @Nonnull + @Override + public CardinalitiesProperty.Cardinalities getCardinalities() { + final int size; + if (collectionValue instanceof AbstractArrayConstructorValue) { + size = Iterables.size(collectionValue.getChildren()); + } else { + return CardinalitiesProperty.Cardinalities.unknownMaxCardinality(); + } + final var cardinality = CardinalitiesProperty.Cardinality.ofCardinality(size); + return new CardinalitiesProperty.Cardinalities(cardinality, cardinality); + } + + @Nullable + @Override + public Object eval(@Nullable final FDBRecordStoreBase store, @Nonnull final EvaluationContext context) { + throw new IllegalStateException("unable to eval an streaming value with eval()"); + } + + @Nonnull + @Override + protected Iterable computeChildren() { + return ImmutableList.of(collectionValue); + } + + @Nonnull + @Override + @SuppressWarnings("PMD.CompareObjectsWithEquals") + public Value withChildren(final Iterable newChildren) { + Verify.verify(Iterables.size(newChildren) == 1); + final Value newCollectionValue = Iterables.getOnlyElement(newChildren); + if (newCollectionValue == collectionValue) { + return this; + } + return withCollectionValue(newCollectionValue); + } + + @Nonnull + @Override + public ExplainTokensWithPrecedence explain(@Nonnull final Iterable> explainSuppliers) { + final var explainTokens = new ExplainTokens().addFunctionCall("explode", + Iterables.getOnlyElement(explainSuppliers).get().getExplainTokens()); + if (sorted) { + explainTokens.addWhitespace().addKeyword("SORTED"); + if (reverse) { + explainTokens.addWhitespace().addKeyword("DESC"); + } + } + return ExplainTokensWithPrecedence.of(explainTokens); + } + + @Override + public int hashCodeWithoutChildren() { + return PlanHashable.objectsPlanHash(PlanHashable.CURRENT_FOR_CONTINUATION, BASE_HASH, preserveOrder, sorted, + reverse); + } + + @Override + public int planHash(@Nonnull final PlanHashMode mode) { + return PlanHashable.objectsPlanHash(mode, BASE_HASH, collectionValue, preserveOrder, sorted, reverse); + } + + @Nonnull + @Override + public ConstrainedBoolean equalsWithoutChildren(@Nonnull final Value other) { + return super.equalsWithoutChildren(other) + .filter(ignored -> { + final var otherExplodeValue = (ExplodeValue)other; + return preserveOrder == otherExplodeValue.preservesOrder() && + sorted == otherExplodeValue.isSorted() && reverse == otherExplodeValue.isReverse(); + }); + } + + @SpotBugsSuppressWarnings("EQ_UNUSUAL") + @SuppressWarnings("EqualsWhichDoesntCheckParameterClass") + @Override + public boolean equals(final Object other) { + return semanticEquals(other, AliasMap.emptyMap()); + } + + @Override + public int hashCode() { + return semanticHashCode(); + } + + @Nonnull + @Override + public PExplodeValue toProto(@Nonnull final PlanSerializationContext serializationContext) { + return PExplodeValue.newBuilder() + .setCollectionValue(collectionValue.toValueProto(serializationContext)) + .setPreserveOrder(preserveOrder) + .setSorted(sorted) + .setReverse(reverse) + .build(); + } + + @Nonnull + @Override + public PValue toValueProto(@Nonnull final PlanSerializationContext serializationContext) { + return PValue.newBuilder().setExplodeValue(toProto(serializationContext)).build(); + } + + @Nonnull + public static ExplodeValue fromProto(@Nonnull final PlanSerializationContext serializationContext, + @Nonnull final PExplodeValue explodeValueProto) { + return new ExplodeValue(Value.fromValueProto(serializationContext, + Objects.requireNonNull(explodeValueProto.getCollectionValue())), explodeValueProto.getPreserveOrder(), + explodeValueProto.getSorted(), explodeValueProto.getReverse()); + } + + /** + * Deserializer. + */ + @AutoService(PlanDeserializer.class) + public static class Deserializer implements PlanDeserializer { + @Nonnull + @Override + public Class getProtoMessageClass() { + return PExplodeValue.class; + } + + @Nonnull + @Override + public ExplodeValue fromProto(@Nonnull final PlanSerializationContext serializationContext, + @Nonnull final PExplodeValue explodeValueProto) { + return ExplodeValue.fromProto(serializationContext, explodeValueProto); + } + } + + /** + * A cursor over the elements of an array, producing a struct of each element and its ordinal. The continuation + * records the position of the next element. + * + *

The continuation is not tied to the array it came from. Resuming is only correct if the array evaluates to the + * same elements in the same order as it did when the continuation was created. + */ + static class Cursor implements RecordCursor { + @Nonnull + private final Executor executor; + + @Nonnull + private final List elements; + + @Nonnull + private final BiFunction, Integer, Object> structCreator; + + private int nextPosition; + + private boolean closed = false; + + Cursor(@Nonnull final Executor executor, @Nonnull final List elements, + @Nonnull final BiFunction, Integer, Object> structCreator, @Nullable final byte[] continuation) { + this(executor, elements, structCreator, + continuation == null ? 0 : Continuation.from(continuation).getNextPosition()); + } + + private Cursor(@Nonnull final Executor executor, @Nonnull final List elements, + @Nonnull final BiFunction, Integer, Object> structCreator, final int nextPosition) { + if (nextPosition < 0 || nextPosition > elements.size()) { + throw new RecordCoreException("invalid continuation position for explode") + .addLogInfo("nextPosition", nextPosition) + .addLogInfo("arraySize", elements.size()); + } + this.executor = executor; + this.elements = elements; + this.structCreator = structCreator; + this.nextPosition = nextPosition; + } + + @Nonnull + @Override + public CompletableFuture> onNext() { + return CompletableFuture.completedFuture(getNext()); + } + + @Nonnull + @Override + public RecordCursorResult getNext() { + if (nextPosition >= elements.size()) { + return RecordCursorResult.exhausted(); + } + final int ordinal = nextPosition; + nextPosition++; + return RecordCursorResult.withNextValue( + QueryResult.ofComputed(structCreator.apply(elements, ordinal)), + new Continuation(nextPosition)); + } + + @Override + public void close() { + closed = true; + } + + @Override + public boolean isClosed() { + return closed; + } + + @Override + public boolean accept(@Nonnull final RecordCursorVisitor visitor) { + visitor.visitEnter(this); + return visitor.visitLeave(this); + } + + @Override + @Nonnull + public Executor getExecutor() { + return executor; + } + + private static class Continuation implements RecordCursorContinuation { + private final int nextPosition; + + Continuation(final int nextPosition) { + this.nextPosition = nextPosition; + } + + @Override + public boolean isEnd() { + return false; + } + + public int getNextPosition() { + return nextPosition; + } + + @Nonnull + @Override + public ByteString toByteString() { + return RecordCursorProto.ExplodeCursorContinuation.newBuilder() + .setNextPosition(nextPosition) + .build() + .toByteString(); + } + + @Nullable + @Override + public byte[] toBytes() { + return toByteString().toByteArray(); + } + + @Nonnull + public static Continuation from(@Nonnull final byte[] unparsedContinuationBytes) { + try { + final var parsed = RecordCursorProto.ExplodeCursorContinuation.parseFrom(unparsedContinuationBytes); + return new Continuation(parsed.getNextPosition()); + } catch (InvalidProtocolBufferException ex) { + throw new RecordCoreException("invalid continuation", ex) + .addLogInfo(LogMessageKeys.RAW_BYTES, ByteArrayUtil2.loggable(unparsedContinuationBytes)); + } + } + } + } +} diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/InExplodeSource.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/InExplodeSource.java new file mode 100644 index 0000000000..56e164c619 --- /dev/null +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/InExplodeSource.java @@ -0,0 +1,200 @@ +/* + * InExplodeSource.java + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2015-2026 Apple Inc. and the FoundationDB project authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.apple.foundationdb.record.query.plan.plans; + +import com.apple.foundationdb.annotation.API; +import com.apple.foundationdb.record.EvaluationContext; +import com.apple.foundationdb.record.ObjectPlanHash; +import com.apple.foundationdb.record.PlanDeserializer; +import com.apple.foundationdb.record.PlanHashable; +import com.apple.foundationdb.record.PlanSerializationContext; +import com.apple.foundationdb.record.RecordCoreException; +import com.apple.foundationdb.record.planprotos.PInExplodeSource; +import com.apple.foundationdb.record.planprotos.PInSource; +import com.apple.foundationdb.record.query.expressions.Comparisons; +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.Value; +import com.apple.foundationdb.record.query.plan.explain.ExplainLevel; +import com.apple.foundationdb.record.query.plan.explain.ExplainTokens; +import com.apple.foundationdb.record.query.plan.explain.ExplainTokensWithPrecedence; +import com.google.auto.service.AutoService; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import java.util.List; +import java.util.Objects; +import java.util.Set; + +/** + * Source of values for an "IN" query that is planned from a quantifier ranging over an {@link ExplodeValue}. Like + * the explode itself, it binds the struct of each element of the collection, so the correlated inner plan can read + * the element and the ordinal the same way it would read them from the explode. + * + *

The source is sorted if the explode is {@linkplain ExplodeValue#isSorted() sorted}, and reverse if the explode + * is {@linkplain ExplodeValue#isReverse() reverse}. + */ +@API(API.Status.INTERNAL) +public class InExplodeSource extends InComparandSource { + @Nonnull + private static final ObjectPlanHash OBJECT_PLAN_HASH_IN_EXPLODE_SOURCE = new ObjectPlanHash("In-Explode"); + + @Nonnull + private final ExplodeValue explodeValue; + + protected InExplodeSource(@Nonnull final PlanSerializationContext serializationContext, + @Nonnull final PInExplodeSource inExplodeSourceProto) { + this(Objects.requireNonNull(inExplodeSourceProto.getSuper()).getBindingName(), + (ExplodeValue)Value.fromValueProto(serializationContext, + Objects.requireNonNull(inExplodeSourceProto.getExplodeValue()))); + } + + public InExplodeSource(@Nonnull final String bindingName, @Nonnull final ExplodeValue explodeValue) { + super(bindingName, new Comparisons.ValueComparison(Comparisons.Type.IN, explodeValue.getCollectionValue())); + this.explodeValue = explodeValue; + } + + @Override + public int planHash(@Nonnull final PlanHashMode mode) { + if (mode.getKind() == PlanHashKind.LEGACY) { + // not every collection value supports the legacy mode + return PlanHashable.objectsPlanHash(mode, baseHash(mode, OBJECT_PLAN_HASH_IN_EXPLODE_SOURCE), getComparison(), + isSorted(), isReverse()); + } + // the comparison does not hash its comparand, so hash the explode, which hashes the collection + return PlanHashable.objectsPlanHash(mode, baseHash(mode, OBJECT_PLAN_HASH_IN_EXPLODE_SOURCE), explodeValue); + } + + @Override + public boolean isSorted() { + return explodeValue.isSorted(); + } + + @Override + public boolean isReverse() { + return explodeValue.isReverse(); + } + + @Nonnull + @Override + public ExplainTokensWithPrecedence explain() { + return ExplainTokensWithPrecedence.of( + new ExplainTokens().addNested(ExplainLevel.ALL_DETAILS, + new Comparisons.ValueComparison(Comparisons.Type.IN, explodeValue.getCollectionValue()) + .explain().getExplainTokens(), "...").addNested(explainSuffix())); + } + + @Override + protected int size(@Nonnull final EvaluationContext context) { + final Object collection = explodeValue.getCollectionValue().eval(null, context); + return collection == null ? 0 : ((List)collection).size(); + } + + /** + * Returns the structs of the elements of the collection. + * + * @param context the evaluation context + * @return the structs of the elements of the collection + */ + @Nonnull + @Override + protected List getValues(@Nullable final EvaluationContext context) { + if (context == null) { + throw new RecordCoreException("values of an in-explode source need an evaluation context"); + } + return explodeValue.evalAsList(null, context); + } + + @Nonnull + @Override + public String toString() { + return getBindingName() + " IN " + explodeValue; + } + + @Override + public boolean equals(final Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + final InExplodeSource inExplodeSource = (InExplodeSource)o; + return super.equals(inExplodeSource) && explodeValue.equals(inExplodeSource.explodeValue); + } + + @Override + public int hashCode() { + return Objects.hash(getComparison(), explodeValue); + } + + @Nonnull + @Override + public Type getResultType() { + return explodeValue.getResultType(); + } + + @Nonnull + @Override + public Set getDynamicTypes() { + return explodeValue.getDynamicTypes(); + } + + @Nonnull + @Override + public PInExplodeSource toProto(@Nonnull final PlanSerializationContext serializationContext) { + return PInExplodeSource.newBuilder() + .setSuper(toInSourceSuperProto(serializationContext)) + .setExplodeValue(explodeValue.toValueProto(serializationContext)) + .build(); + } + + @Nonnull + @Override + protected PInSource toInSourceProto(@Nonnull final PlanSerializationContext serializationContext) { + return PInSource.newBuilder().setInExplodeSource(toProto(serializationContext)).build(); + } + + @Nonnull + public static InExplodeSource fromProto(@Nonnull final PlanSerializationContext serializationContext, + @Nonnull final PInExplodeSource inExplodeSourceProto) { + return new InExplodeSource(serializationContext, inExplodeSourceProto); + } + + /** + * Deserializer. + */ + @AutoService(PlanDeserializer.class) + public static class Deserializer implements PlanDeserializer { + @Nonnull + @Override + public Class getProtoMessageClass() { + return PInExplodeSource.class; + } + + @Nonnull + @Override + public InExplodeSource fromProto(@Nonnull final PlanSerializationContext serializationContext, + @Nonnull final PInExplodeSource inExplodeSourceProto) { + return InExplodeSource.fromProto(serializationContext, inExplodeSourceProto); + } + } +} diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/RecordQueryInUnionPlan.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/RecordQueryInUnionPlan.java index 85eedba180..b9bec62110 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/RecordQueryInUnionPlan.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/RecordQueryInUnionPlan.java @@ -49,6 +49,7 @@ import com.apple.foundationdb.record.query.plan.cascades.expressions.AbstractRelationalExpressionWithChildren; import com.apple.foundationdb.record.query.plan.cascades.expressions.RelationalExpression; import com.apple.foundationdb.record.query.plan.cascades.explain.ExplainPlanVisitor; +import com.apple.foundationdb.record.query.plan.cascades.typing.Type; import com.apple.foundationdb.record.query.plan.cascades.values.Value; import com.google.common.base.Verify; import com.google.common.collect.ImmutableList; @@ -202,6 +203,15 @@ public Value getResultValue() { return inner.getFlowedObjectValue(); } + @Nonnull + @Override + protected Set computeDynamicTypes() { + final ImmutableSet.Builder resultBuilder = ImmutableSet.builder(); + resultBuilder.addAll(super.computeDynamicTypes()); + getInSources().forEach(inSource -> resultBuilder.addAll(inSource.getDynamicTypes())); + return resultBuilder.build(); + } + @Nonnull @Override public Set computeCorrelatedTo() { diff --git a/fdb-record-layer-core/src/main/proto/record_cursor.proto b/fdb-record-layer-core/src/main/proto/record_cursor.proto index 5cb54a5ac3..1abb38cdff 100644 --- a/fdb-record-layer-core/src/main/proto/record_cursor.proto +++ b/fdb-record-layer-core/src/main/proto/record_cursor.proto @@ -183,6 +183,10 @@ message RangeCursorContinuation { optional int64 nextPosition = 1; } +message ExplodeCursorContinuation { + optional int32 nextPosition = 1; +} + message KeyValueCursorContinuation { optional bytes inner_continuation = 1; optional sfixed64 magic_number = 2; diff --git a/fdb-record-layer-core/src/main/proto/record_query_plan.proto b/fdb-record-layer-core/src/main/proto/record_query_plan.proto index 52a61a6936..475cccda32 100644 --- a/fdb-record-layer-core/src/main/proto/record_query_plan.proto +++ b/fdb-record-layer-core/src/main/proto/record_query_plan.proto @@ -287,6 +287,7 @@ message PValue { PCardinalityValue cardinality_value = 63; PArrayAggValue array_agg_value = 64; PCurrentTimestampValue current_timestamp_value = 65; + PExplodeValue explode_value = 66; } // Removed types that are no longer supported reserved 38; @@ -1409,6 +1410,13 @@ message PRangeValue { optional PValue step_child = 3; } +message PExplodeValue { + optional PValue collection_value = 1; + optional bool preserve_order = 2; + optional bool sorted = 3; + optional bool reverse = 4; +} + message PArrayDistinctValue { optional PValue child_value = 1; } @@ -2057,6 +2065,7 @@ message PInSource { PSortedInParameterSource sorted_in_parameter_source = 4; PInValuesSource in_values_source = 5; PSortedInValuesSource sorted_in_values_source = 6; + PInExplodeSource in_explode_source = 7; } } @@ -2090,6 +2099,12 @@ message PSortedInValuesSource { optional bool reverse = 2; } +message PInExplodeSource { + // the comparison of the source is derived from the explode, so it is not serialized + optional PInSource.Super super = 1; + optional PValue explode_value = 2; +} + // // PRecordQueryInParameterJoinPlan // diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/query/plan/cascades/values/ExplodeValueTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/query/plan/cascades/values/ExplodeValueTest.java new file mode 100644 index 0000000000..d3055e16a8 --- /dev/null +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/query/plan/cascades/values/ExplodeValueTest.java @@ -0,0 +1,507 @@ +/* + * ExplodeValueTest.java + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2015-2026 Apple Inc. and the FoundationDB project authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.apple.foundationdb.record.query.plan.cascades.values; + +import com.apple.foundationdb.record.Bindings; +import com.apple.foundationdb.record.EvaluationContext; +import com.apple.foundationdb.record.ExecuteProperties; +import com.apple.foundationdb.record.PlanHashable; +import com.apple.foundationdb.record.PlanSerializationContext; +import com.apple.foundationdb.record.RecordCoreException; +import com.apple.foundationdb.record.RecordCursor; +import com.apple.foundationdb.record.RecordCursorResult; +import com.apple.foundationdb.record.RecordCursorVisitor; +import com.apple.foundationdb.record.TestRecords4Proto; +import com.apple.foundationdb.record.provider.foundationdb.FDBRecordStoreBase; +import com.apple.foundationdb.record.query.plan.cascades.CorrelationIdentifier; +import com.apple.foundationdb.record.query.plan.cascades.properties.DerivationsProperty; +import com.apple.foundationdb.record.query.plan.cascades.typing.Type; +import com.apple.foundationdb.record.query.plan.cascades.typing.TypeRepository; +import com.apple.foundationdb.record.query.plan.explain.DefaultExplainFormatter; +import com.apple.foundationdb.record.query.plan.explain.DefaultExplainSymbolMap; +import com.apple.foundationdb.record.query.plan.plans.QueryResult; +import com.apple.foundationdb.record.query.plan.plans.RecordQueryTableFunctionPlan; +import com.apple.foundationdb.record.query.plan.serialization.DefaultPlanSerializationRegistry; +import com.apple.test.BooleanSource; +import com.google.common.base.VerifyException; +import com.google.common.collect.ImmutableList; +import com.google.protobuf.Message; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.mockito.Mockito; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Tests for {@link ExplodeValue}. + */ +class ExplodeValueTest { + + private static final CorrelationIdentifier RESTAURANT_ALIAS = CorrelationIdentifier.of("restaurant"); + + @Nonnull + private static final FDBRecordStoreBase STORE = mockStore(); + + @Nonnull + @SuppressWarnings("unchecked") + private static FDBRecordStoreBase mockStore() { + final FDBRecordStoreBase store = Mockito.mock(FDBRecordStoreBase.class); + Mockito.when(store.getExecutor()).thenReturn(ForkJoinPool.commonPool()); + return store; + } + + @Nonnull + private static PlanSerializationContext newSerializationContext() { + return new PlanSerializationContext(new DefaultPlanSerializationRegistry(), + PlanHashable.CURRENT_FOR_CONTINUATION); + } + + @Nonnull + private static TestRecords4Proto.RestaurantReview review(final long reviewer, final int rating) { + return TestRecords4Proto.RestaurantReview.newBuilder().setReviewer(reviewer).setRating(rating).build(); + } + + @Nonnull + private static Value reviewsOfRestaurant() { + return FieldValue.ofFieldName(QuantifiedObjectValue.of(RESTAURANT_ALIAS, + Type.Record.fromDescriptor(TestRecords4Proto.RestaurantRecord.getDescriptor())), "reviews"); + } + + @Nonnull + private static EvaluationContext contextFor(@Nonnull final ExplodeValue value) { + return EvaluationContext.forTypeRepository(TypeRepository.newBuilder().addAllTypes(value.getDynamicTypes()).build()); + } + + @Nonnull + private static EvaluationContext contextFor(@Nonnull final ExplodeValue value, + @Nonnull final TestRecords4Proto.RestaurantRecord restaurant) { + return contextFor(value).withBinding(Bindings.Internal.CORRELATION, RESTAURANT_ALIAS, + QueryResult.ofComputed(restaurant)); + } + + /** + * Drains the stream of an explode, returning each struct it produces as the list of its fields, that is + * {@code [element, ordinal]}. + */ + @Nonnull + private static List> explode(@Nonnull final ExplodeValue value, @Nonnull final EvaluationContext context, + @Nullable final byte[] continuation, + @Nonnull final ExecuteProperties executeProperties) { + final var structs = value.evalAsStream(STORE, context, continuation, executeProperties).asList().join(); + return structs.stream() + .map(result -> fieldsOf(Objects.requireNonNull(result.getDatum()))) + .collect(ImmutableList.toImmutableList()); + } + + /** + * Evaluates an explode eagerly, returning each struct it produces as the list of its fields. + */ + @Nonnull + private static List> explodeAsList(@Nonnull final ExplodeValue value, @Nonnull final EvaluationContext context) { + return value.evalAsList(null, context).stream() + .map(ExplodeValueTest::fieldsOf) + .collect(ImmutableList.toImmutableList()); + } + + @Nonnull + private static List fieldsOf(@Nonnull final Object struct) { + final var message = (Message)struct; + return message.getDescriptorForType().getFields().stream() + .map(message::getField) + .collect(ImmutableList.toImmutableList()); + } + + @Nonnull + private static List> explode(@Nonnull final ExplodeValue value, @Nonnull final EvaluationContext context) { + return explode(value, context, null, ExecuteProperties.SERIAL_EXECUTE); + } + + @ParameterizedTest + @BooleanSource + void resultTypeIsTheElementAndItsOrdinal(final boolean preserveOrder) { + final var value = new ExplodeValue(LiteralValue.ofList(List.of("a", "b")), preserveOrder); + Assertions.assertEquals(preserveOrder, value.preservesOrder()); + + final var fields = value.getResultType().getFields(); + Assertions.assertEquals(2, fields.size()); + Assertions.assertEquals(((Type.Array)value.getCollectionValue().getResultType()).getElementType(), fields.get(0).getFieldType()); + Assertions.assertEquals(Type.primitiveType(Type.TypeCode.INT, false), fields.get(1).getFieldType()); + // Both fields are declared anonymous, the way an explode WITH ORDINALITY declares its struct, and are so named + // by their position. Preserving the order does not change the shape. + Assertions.assertEquals("_0", fields.get(0).getFieldName()); + Assertions.assertEquals("_1", fields.get(1).getFieldName()); + + // The struct type is created by the value, so it is one of its dynamic types. + Assertions.assertTrue(value.getDynamicTypes().contains(value.getResultType())); + } + + @ParameterizedTest + @BooleanSource + void producesEveryElementWithItsZeroBasedOrdinal(final boolean preserveOrder) { + final var value = new ExplodeValue(LiteralValue.ofList(List.of("a", "b", "c")), preserveOrder); + Assertions.assertEquals(List.of(List.of("a", 0), List.of("b", 1), List.of("c", 2)), + explode(value, contextFor(value))); + } + + @Test + void producesOrdinalsEvenWhenTheElementsAreOfTheOrdinalType() { + final var value = new ExplodeValue(LiteralValue.ofList(List.of(7, 7, 9)), true); + Assertions.assertEquals(List.of(List.of(7, 0), List.of(7, 1), List.of(9, 2)), + explode(value, contextFor(value))); + } + + @Test + void explodesAnArrayOfMessagesOfACorrelatedRecord() { + final var value = new ExplodeValue(reviewsOfRestaurant(), true); + final var restaurant = TestRecords4Proto.RestaurantRecord.newBuilder() + .setRestNo(1L) + .addReviews(review(10L, 5)) + .addReviews(review(20L, 3)) + .build(); + + final var exploded = explode(value, contextFor(value, restaurant)); + Assertions.assertEquals(2, exploded.size()); + final var firstReview = (Message)exploded.get(0).get(0); + final var secondReview = (Message)exploded.get(1).get(0); + Assertions.assertEquals(10L, firstReview.getField(firstReview.getDescriptorForType().findFieldByName("reviewer"))); + Assertions.assertEquals(20L, secondReview.getField(secondReview.getDescriptorForType().findFieldByName("reviewer"))); + Assertions.assertEquals(List.of(0, 1), List.of(exploded.get(0).get(1), exploded.get(1).get(1))); + } + + @ParameterizedTest + @BooleanSource + void evaluatingAsAListProducesWhatTheStreamProduces(final boolean preserveOrder) { + final var literal = new ExplodeValue(LiteralValue.ofList(List.of("a", "b", "c")), preserveOrder); + final var literalContext = contextFor(literal); + Assertions.assertEquals(explode(literal, literalContext), explodeAsList(literal, literalContext)); + + final var reviews = new ExplodeValue(reviewsOfRestaurant(), preserveOrder); + final var restaurant = TestRecords4Proto.RestaurantRecord.newBuilder() + .setRestNo(1L) + .addReviews(review(10L, 5)) + .addReviews(review(20L, 3)) + .build(); + // the same context for both, so that the messages of the reviews share their descriptor + final var reviewsContext = contextFor(reviews, restaurant); + Assertions.assertEquals(explode(reviews, reviewsContext), explodeAsList(reviews, reviewsContext)); + } + + @Test + void evaluatingAsAListOfAnEmptyOrNullArrayProducesNothing() { + final var empty = new ExplodeValue(new LiteralValue<>(new Type.Array(Type.primitiveType(Type.TypeCode.INT, false)), List.of()), true); + Assertions.assertEquals(List.of(), explodeAsList(empty, contextFor(empty))); + + final var nullArray = new ExplodeValue(new NullValue(new Type.Array(Type.primitiveType(Type.TypeCode.INT, false))), true); + Assertions.assertEquals(List.of(), explodeAsList(nullArray, contextFor(nullArray))); + } + + @ParameterizedTest + @BooleanSource + void sortedExplodeProducesTheElementsSortedAndNumbersThemInThatOrder(final boolean reverse) { + final var value = new ExplodeValue(LiteralValue.ofList(List.of(30, 10, 20)), false, true, reverse); + Assertions.assertEquals(reverse + ? List.of(List.of(30, 0), List.of(20, 1), List.of(10, 2)) + : List.of(List.of(10, 0), List.of(20, 1), List.of(30, 2)), + explodeAsList(value, contextFor(value))); + } + + @Test + void anEmptyOrNullArrayProducesNothing() { + final var value = new ExplodeValue(reviewsOfRestaurant()); + final var restaurantWithoutReviews = TestRecords4Proto.RestaurantRecord.newBuilder().setRestNo(1L).build(); + Assertions.assertEquals(List.of(), explode(value, contextFor(value, restaurantWithoutReviews))); + + final var nullArray = new ExplodeValue(new NullValue(new Type.Array(Type.primitiveType(Type.TypeCode.INT, false)))); + Assertions.assertEquals(List.of(), explode(nullArray, contextFor(nullArray))); + } + + @Test + void honorsSkipAndLimit() { + final var value = new ExplodeValue(LiteralValue.ofList(List.of("a", "b", "c", "d")), true); + final var executeProperties = ExecuteProperties.newBuilder().setSkip(1).setReturnedRowLimit(2).build(); + Assertions.assertEquals(List.of(List.of("b", 1), List.of("c", 2)), + explode(value, contextFor(value), null, executeProperties)); + } + + @Test + void resumesFromAContinuation() { + final var value = new ExplodeValue(LiteralValue.ofList(List.of("a", "b", "c")), true); + final var context = contextFor(value); + + final RecordCursorResult first = + value.evalAsStream(STORE, context, null, ExecuteProperties.SERIAL_EXECUTE).getNext(); + Assertions.assertTrue(first.hasNext()); + + // The rest of the stream picks up after the first element, ordinals included. + Assertions.assertEquals(List.of(List.of("b", 1), List.of("c", 2)), + explode(value, context, first.getContinuation().toBytes(), ExecuteProperties.SERIAL_EXECUTE)); + } + + @Test + void theContinuationOfTheLastElementResumesToNothing() { + final var value = new ExplodeValue(LiteralValue.ofList(List.of("a", "b"))); + final var context = contextFor(value); + final var cursor = value.evalAsStream(STORE, context, null, ExecuteProperties.SERIAL_EXECUTE); + cursor.getNext(); + final RecordCursorResult last = cursor.getNext(); + Assertions.assertTrue(last.hasNext()); + + // A continuation handed out with a value is never an end continuation, even after the last element. + Assertions.assertFalse(last.getContinuation().isEnd()); + Assertions.assertEquals(List.of(), + explode(value, context, last.getContinuation().toBytes(), ExecuteProperties.SERIAL_EXECUTE)); + Assertions.assertTrue(cursor.getNext().getContinuation().isEnd()); + } + + @Test + void rejectsAContinuationPastTheEndOfTheArray() { + final var longer = new ExplodeValue(LiteralValue.ofList(List.of("a", "b", "c"))); + final var cursor = longer.evalAsStream(STORE, contextFor(longer), null, ExecuteProperties.SERIAL_EXECUTE); + cursor.getNext(); + cursor.getNext(); + final byte[] continuation = cursor.getNext().getContinuation().toBytes(); + + // Resuming the same position over an array that has since shrunk is an error. + final var shorter = new ExplodeValue(LiteralValue.ofList(List.of("a"))); + Assertions.assertThrows(RecordCoreException.class, + () -> shorter.evalAsStream(STORE, contextFor(shorter), continuation, ExecuteProperties.SERIAL_EXECUTE)); + } + + @Test + void rejectsAContinuationThatCannotBeParsed() { + final var value = new ExplodeValue(LiteralValue.ofList(List.of("a"))); + Assertions.assertThrows(RecordCoreException.class, + () -> value.evalAsStream(STORE, contextFor(value), new byte[] {(byte)0xff, (byte)0xff}, + ExecuteProperties.SERIAL_EXECUTE)); + } + + @Test + void createsAStructOnlyOnceTheCursorReachesItsElement() { + final var structsCreated = new AtomicInteger(); + final var cursor = new ExplodeValue.Cursor(ForkJoinPool.commonPool(), List.of("a", "b", "c"), + (array, ordinal) -> { + structsCreated.incrementAndGet(); + return List.of(array.get(ordinal), ordinal); + }, null); + Assertions.assertEquals(0, structsCreated.get()); + + Assertions.assertEquals(List.of("a", 0), Objects.requireNonNull(cursor.getNext().get()).getDatum()); + Assertions.assertEquals(1, structsCreated.get()); + } + + @Test + void theCardinalityOfAnExplodeOfAnArrayConstructorIsItsNumberOfElements() { + final var value = new ExplodeValue(AbstractArrayConstructorValue.LightArrayConstructorValue.of( + List.of(LiteralValue.ofScalar(1), LiteralValue.ofScalar(2), LiteralValue.ofScalar(3)))); + final var cardinalities = value.getCardinalities(); + Assertions.assertEquals(3, cardinalities.getMinCardinality().getCardinality()); + Assertions.assertEquals(3, cardinalities.getMaxCardinality().getCardinality()); + } + + @Test + void theMaximumCardinalityOfAnyOtherExplodeIsUnknown() { + Assertions.assertTrue(new ExplodeValue(reviewsOfRestaurant()).getCardinalities().getMaxCardinality().isUnknown()); + } + + @Test + void theCursorIsALeafOfTheCursorTree() { + final var cursor = new ExplodeValue.Cursor(ForkJoinPool.commonPool(), List.of("a"), + (array, ordinal) -> array.get(ordinal), null); + final var entered = new AtomicInteger(); + final var left = new AtomicInteger(); + final boolean result = cursor.accept(new RecordCursorVisitor() { + @Override + public boolean visitEnter(final RecordCursor visited) { + Assertions.assertSame(cursor, visited); + entered.incrementAndGet(); + return true; + } + + @Override + public boolean visitLeave(final RecordCursor visited) { + Assertions.assertSame(cursor, visited); + left.incrementAndGet(); + return true; + } + }); + Assertions.assertTrue(result); + Assertions.assertEquals(1, entered.get()); + Assertions.assertEquals(1, left.get()); + } + + @Test + void rejectsACollectionThatIsNotAnArray() { + Assertions.assertThrows(VerifyException.class, () -> new ExplodeValue(LiteralValue.ofScalar(1))); + } + + @Test + void cannotBeEvaluatedAsAScalar() { + final var value = new ExplodeValue(LiteralValue.ofList(List.of(1, 2))); + Assertions.assertThrows(IllegalStateException.class, () -> value.eval(null, contextFor(value))); + } + + @Test + void equalityAndHashingFollowTheCollection() { + final var collection = LiteralValue.ofList(List.of(1, 2)); + final var value = new ExplodeValue(collection); + + Assertions.assertEquals(value, new ExplodeValue(LiteralValue.ofList(List.of(1, 2)))); + Assertions.assertEquals(value.hashCode(), new ExplodeValue(LiteralValue.ofList(List.of(1, 2))).hashCode()); + Assertions.assertNotEquals(value, new ExplodeValue(LiteralValue.ofList(List.of(1, 3)))); + Assertions.assertNotEquals(value.planHash(PlanHashable.CURRENT_FOR_CONTINUATION), + new ExplodeValue(LiteralValue.ofList(List.of(1, 3))).planHash(PlanHashable.CURRENT_FOR_CONTINUATION)); + + Assertions.assertSame(value, value.withChildren(List.of(collection))); + final var otherCollection = LiteralValue.ofList(List.of(3, 4)); + Assertions.assertEquals(new ExplodeValue(otherCollection), value.withChildren(List.of(otherCollection))); + } + + @Test + void equalityAndHashingFollowWhetherTheOrderIsPreserved() { + final var collection = LiteralValue.ofList(List.of(1, 2)); + final var value = new ExplodeValue(collection); + final var ordered = new ExplodeValue(collection, true); + + Assertions.assertNotEquals(value, ordered); + Assertions.assertNotEquals(value.planHash(PlanHashable.CURRENT_FOR_CONTINUATION), + ordered.planHash(PlanHashable.CURRENT_FOR_CONTINUATION)); + + // rebasing the collection keeps preserving the order + final var otherCollection = LiteralValue.ofList(List.of(3, 4)); + Assertions.assertEquals(new ExplodeValue(otherCollection, true), ordered.withChildren(List.of(otherCollection))); + Assertions.assertEquals(new ExplodeValue(otherCollection, true), ordered.withCollectionValue(otherCollection)); + } + + @Test + void equalityAndHashingFollowTheSortOrder() { + final var collection = LiteralValue.ofList(List.of(1, 2)); + final var value = new ExplodeValue(collection); + final var ascending = value.sorted(false); + final var descending = value.sorted(true); + + Assertions.assertNotEquals(value, ascending); + Assertions.assertNotEquals(ascending, descending); + Assertions.assertEquals(new ExplodeValue(collection, false, true, true), descending); + Assertions.assertNotEquals(ascending.planHash(PlanHashable.CURRENT_FOR_CONTINUATION), + descending.planHash(PlanHashable.CURRENT_FOR_CONTINUATION)); + + // rebasing the collection keeps the sort order + final var otherCollection = LiteralValue.ofList(List.of(3, 4)); + Assertions.assertEquals(new ExplodeValue(otherCollection, false, true, true), + descending.withChildren(List.of(otherCollection))); + } + + @Test + void rejectsSortingAnExplodeThatPreservesItsOrder() { + final var collection = LiteralValue.ofList(List.of(1, 2)); + Assertions.assertThrows(VerifyException.class, () -> new ExplodeValue(collection, true, true, false)); + Assertions.assertThrows(VerifyException.class, () -> new ExplodeValue(collection, true).sorted(false)); + } + + @Test + void rejectsReversingAnExplodeThatIsNotSorted() { + Assertions.assertThrows(VerifyException.class, + () -> new ExplodeValue(LiteralValue.ofList(List.of(1, 2)), false, false, true)); + } + + @Test + void rejectsSortingElementsThatCannotBeSorted() { + final var records = new LiteralValue<>(new Type.Array(Type.Record.fromFields(List.of( + Type.Record.Field.of(Type.primitiveType(Type.TypeCode.LONG), Optional.of("a"))))), List.of()); + Assertions.assertThrows(VerifyException.class, () -> new ExplodeValue(records).sorted(false)); + final var bytes = new LiteralValue<>(new Type.Array(Type.primitiveType(Type.TypeCode.BYTES, false)), List.of()); + Assertions.assertThrows(VerifyException.class, () -> new ExplodeValue(bytes).sorted(false)); + // an explode of such elements that is not sorted is fine + Assertions.assertFalse(new ExplodeValue(records).isSorted()); + } + + @Test + void explainsAsAnExplodeOfTheCollection() { + final var value = new ExplodeValue(LiteralValue.ofList(List.of(1, 2))); + Assertions.assertEquals("explode([1, 2])", + value.explain().getExplainTokens().render(DefaultExplainFormatter.create(DefaultExplainSymbolMap::new)).toString()); + } + + @ParameterizedTest + @BooleanSource + void explainsTheSortOrder(final boolean reverse) { + final var value = new ExplodeValue(LiteralValue.ofList(List.of(1, 2))).sorted(reverse); + Assertions.assertEquals(reverse ? "explode([1, 2]) SORTED DESC" : "explode([1, 2]) SORTED", + value.explain().getExplainTokens().render(DefaultExplainFormatter.create(DefaultExplainSymbolMap::new)).toString()); + } + + @ParameterizedTest + @BooleanSource + void derivationsOfATableFunctionKeepTheCollectionCorrelationAndTheStruct(final boolean preserveOrder) { + final var sourceAlias = CorrelationIdentifier.of("source"); + final var arrayType = new Type.Array(false, Type.primitiveType(Type.TypeCode.INT, false)); + final var sourceType = Type.Record.fromFields(List.of(Type.Record.Field.of(arrayType, Optional.of("arr")))); + final Value collectionValue = FieldValue.ofFieldName(QuantifiedObjectValue.of(sourceAlias, sourceType), "arr"); + final var explodeValue = new ExplodeValue(collectionValue, preserveOrder); + + final var derivations = new DerivationsProperty.DerivationsVisitor() + .visitTableFunctionPlan(new RecordQueryTableFunctionPlan(explodeValue)); + + Assertions.assertEquals(1, derivations.getResultValues().size()); + final Value derived = derivations.getResultValues().get(0); + // the representative of some struct of the explode is still correlated to what the collection is correlated to + Assertions.assertTrue(derived.getCorrelatedTo().contains(sourceAlias)); + // and it has the shape of the struct the explode flows: the first element of the collection, and the ordinal + Assertions.assertEquals(explodeValue.getResultType(), derived.getResultType()); + Assertions.assertTrue(derived instanceof RecordConstructorValue); + final List columns = ImmutableList.copyOf(derived.getChildren()); + Assertions.assertEquals(2, columns.size()); + Assertions.assertTrue(columns.get(0) instanceof FirstOrDefaultValue); + } + + @ParameterizedTest + @BooleanSource + void protoRoundTrip(final boolean preserveOrder) { + final var value = new ExplodeValue(reviewsOfRestaurant(), preserveOrder); + final var valueProto = value.toValueProto(newSerializationContext()); + Assertions.assertTrue(valueProto.hasExplodeValue()); + + final var deserialized = Value.fromValueProto(newSerializationContext(), valueProto); + Assertions.assertInstanceOf(ExplodeValue.class, deserialized); + Assertions.assertEquals(value, deserialized); + Assertions.assertEquals(value.getResultType(), deserialized.getResultType()); + Assertions.assertEquals(preserveOrder, ((ExplodeValue)deserialized).preservesOrder()); + } + + @ParameterizedTest + @BooleanSource + void protoRoundTripOfASortedExplode(final boolean reverse) { + final var value = new ExplodeValue(AbstractArrayConstructorValue.LightArrayConstructorValue.of( + List.of(LiteralValue.ofScalar(30), LiteralValue.ofScalar(10), LiteralValue.ofScalar(20)))).sorted(reverse); + final var deserialized = Value.fromValueProto(newSerializationContext(), value.toValueProto(newSerializationContext())); + Assertions.assertEquals(value, deserialized); + Assertions.assertTrue(((ExplodeValue)deserialized).isSorted()); + Assertions.assertEquals(reverse, ((ExplodeValue)deserialized).isReverse()); + Assertions.assertEquals(explodeAsList(value, contextFor(value)), explodeAsList((ExplodeValue)deserialized, contextFor(value))); + } +} diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/query/plan/plans/InExplodeSourceTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/query/plan/plans/InExplodeSourceTest.java new file mode 100644 index 0000000000..89d6ffc4bd --- /dev/null +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/query/plan/plans/InExplodeSourceTest.java @@ -0,0 +1,156 @@ +/* + * InExplodeSourceTest.java + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2015-2026 Apple Inc. and the FoundationDB project authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.apple.foundationdb.record.query.plan.plans; + +import com.apple.foundationdb.record.EvaluationContext; +import com.apple.foundationdb.record.PlanHashable; +import com.apple.foundationdb.record.PlanSerializationContext; +import com.apple.foundationdb.record.RecordCoreException; +import com.apple.foundationdb.record.query.plan.cascades.typing.Type; +import com.apple.foundationdb.record.query.plan.cascades.typing.TypeRepository; +import com.apple.foundationdb.record.query.plan.cascades.values.AbstractArrayConstructorValue; +import com.apple.foundationdb.record.query.plan.cascades.values.ExplodeValue; +import com.apple.foundationdb.record.query.plan.cascades.values.LiteralValue; +import com.apple.foundationdb.record.query.plan.cascades.values.ParameterObjectValue; +import com.apple.foundationdb.record.query.plan.cascades.values.Value; +import com.apple.foundationdb.record.query.plan.explain.DefaultExplainFormatter; +import com.apple.foundationdb.record.query.plan.explain.DefaultExplainSymbolMap; +import com.apple.foundationdb.record.query.plan.serialization.DefaultPlanSerializationRegistry; +import com.google.common.collect.ImmutableList; +import com.google.protobuf.Message; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import javax.annotation.Nonnull; +import java.util.List; +import java.util.stream.Stream; + +/** + * Tests for {@link InExplodeSource} and the values it is built from, in particular that they survive serialization, + * so that a plan using them can be executed by a server that did not plan it. + */ +class InExplodeSourceTest { + + @Nonnull + private static PlanSerializationContext newSerializationContext() { + return new PlanSerializationContext(new DefaultPlanSerializationRegistry(), + PlanHashable.CURRENT_FOR_CONTINUATION); + } + + @Nonnull + private static Value elements() { + return AbstractArrayConstructorValue.LightArrayConstructorValue.of( + List.of(LiteralValue.ofScalar(30L), LiteralValue.ofScalar(10L), LiteralValue.ofScalar(20L))); + } + + @Nonnull + private static List elementsOf(@Nonnull final InExplodeSource source) { + final var context = EvaluationContext.forTypeRepository( + TypeRepository.newBuilder().addAllTypes(source.getDynamicTypes()).build()); + return source.getValues(context).stream() + .map(struct -> { + final var message = (Message)struct; + return message.getField(message.getDescriptorForType().getFields().get(0)); + }) + .collect(ImmutableList.toImmutableList()); + } + + @Nonnull + static Stream sources() { + return Stream.of( + Arguments.of(new InExplodeSource("in", new ExplodeValue(elements())), List.of(30L, 10L, 20L)), + Arguments.of(new InExplodeSource("in", new ExplodeValue(elements()).sorted(false)), List.of(10L, 20L, 30L)), + Arguments.of(new InExplodeSource("in", new ExplodeValue(elements()).sorted(true)), List.of(30L, 20L, 10L))); + } + + @ParameterizedTest + @MethodSource("sources") + void producesTheElementsOfTheExplode(@Nonnull final InExplodeSource source, @Nonnull final List expected) { + Assertions.assertEquals(expected, elementsOf(source)); + } + + @ParameterizedTest + @MethodSource("sources") + void sizeCountsTheElementsWithoutTheirType(@Nonnull final InExplodeSource source, @Nonnull final List expected) { + // the empty context has no type repository, so creating any struct would fail + Assertions.assertEquals(expected.size(), source.size(EvaluationContext.EMPTY)); + } + + @Test + void sizeOfANullCollectionIsZero() { + final var source = new InExplodeSource("in", + new ExplodeValue(ParameterObjectValue.of("p", new Type.Array(Type.primitiveType(Type.TypeCode.LONG, false))))); + Assertions.assertEquals(0, source.size(EvaluationContext.EMPTY.withBinding("p", null))); + Assertions.assertEquals(2, source.size(EvaluationContext.EMPTY.withBinding("p", List.of(1L, 2L)))); + } + + @ParameterizedTest + @MethodSource("sources") + void flowsTheStructOfTheExplode(@Nonnull final InExplodeSource source, @Nonnull final List expected) { + final var explodeValue = new ExplodeValue(elements()); + Assertions.assertEquals(explodeValue.getResultType(), source.getResultType()); + } + + @ParameterizedTest + @MethodSource("sources") + void equalSourcesHaveEqualHashCodes(@Nonnull final InExplodeSource source, @Nonnull final List expected) { + final var deserialized = InSource.fromInSourceProto(newSerializationContext(), + source.toInSourceProto(newSerializationContext())); + Assertions.assertEquals(source, deserialized); + Assertions.assertEquals(source.hashCode(), deserialized.hashCode()); + Assertions.assertNotEquals(source, new InExplodeSource("other", new ExplodeValue(elements()))); + } + + @ParameterizedTest + @MethodSource("sources") + void explainsTheCollectionAndTheSortOrder(@Nonnull final InExplodeSource source, @Nonnull final List expected) { + final String explained = source.explain().getExplainTokens() + .render(DefaultExplainFormatter.create(DefaultExplainSymbolMap::new)).toString(); + Assertions.assertTrue(explained.startsWith("IN "), explained); + Assertions.assertEquals(source.isSorted(), explained.contains("SORTED"), explained); + Assertions.assertEquals(source.isReverse(), explained.contains("DESC"), explained); + } + + @Test + void valuesNeedAnEvaluationContext() { + final var source = new InExplodeSource("in", new ExplodeValue(elements())); + Assertions.assertThrows(RecordCoreException.class, source::getValues); + } + + @ParameterizedTest + @MethodSource("sources") + void protoRoundTrip(@Nonnull final InExplodeSource source, @Nonnull final List expected) { + final var inSourceProto = source.toInSourceProto(newSerializationContext()); + Assertions.assertTrue(inSourceProto.hasInExplodeSource()); + + final var deserialized = InSource.fromInSourceProto(newSerializationContext(), inSourceProto); + Assertions.assertInstanceOf(InExplodeSource.class, deserialized); + Assertions.assertEquals(source, deserialized); + Assertions.assertEquals(source.planHash(PlanHashable.CURRENT_FOR_CONTINUATION), + deserialized.planHash(PlanHashable.CURRENT_FOR_CONTINUATION)); + Assertions.assertEquals(source.isSorted(), deserialized.isSorted()); + Assertions.assertEquals(source.isReverse(), deserialized.isReverse()); + Assertions.assertEquals(expected, elementsOf((InExplodeSource)deserialized)); + } +}