Merge branch 'main' into udf

Author: pengpeng-lu

.github/workflows/nightly.yml

@@ -3,6 +3,7 @@ name: Nightly
 on:
   schedule:
     - cron: "0 7 * * 1-5"
+  workflow_dispatch:
 
 jobs:
   gradle:

build.gradle

@@ -279,6 +279,7 @@ subprojects {
                 sourceSets["test"].resources.srcDirs.stream().flatMap( resourceDir ->
                     fileTree(dir: resourceDir).matching {
                         include '**/*.yamsql'
+                        exclude '/initial-version/*'
                         exclude '/supported-version/*'
                     }.stream()
                 ).forEach { yamsql ->

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/RecordQueryStreamingAggregationPlan.java

@@ -35,7 +35,6 @@
 import com.apple.foundationdb.record.provider.common.StoreTimer;
 import com.apple.foundationdb.record.provider.foundationdb.FDBRecordStoreBase;
 import com.apple.foundationdb.record.provider.foundationdb.FDBStoreTimer;
-import com.apple.foundationdb.record.query.plan.explain.ExplainPlanVisitor;
 import com.apple.foundationdb.record.query.plan.cascades.AliasMap;
 import com.apple.foundationdb.record.query.plan.cascades.CorrelationIdentifier;
 import com.apple.foundationdb.record.query.plan.cascades.Quantifier;
@@ -50,6 +49,7 @@
 import com.apple.foundationdb.record.query.plan.cascades.values.ObjectValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.Value;
 import com.apple.foundationdb.record.query.plan.cascades.values.translation.TranslationMap;
+import com.apple.foundationdb.record.query.plan.explain.ExplainPlanVisitor;
 import com.apple.foundationdb.record.query.plan.serialization.PlanSerialization;
 import com.google.auto.service.AutoService;
 import com.google.common.collect.ImmutableList;
@@ -370,7 +370,8 @@ public PRecordQueryStreamingAggregationPlan toProto(@Nonnull final PlanSerializa
         }
         builder.setGroupingKeyAlias(groupingKeyAlias.getId())
                 .setAggregateAlias(aggregateAlias.getId())
-                .setCompleteResultValue(completeResultValue.toValueProto(serializationContext));
+                .setCompleteResultValue(completeResultValue.toValueProto(serializationContext))
+                .setIsCreateDefaultOnEmpty(isCreateDefaultOnEmpty);
         return builder.build();
     }
 
@@ -384,15 +385,18 @@ public PRecordQueryPlan toRecordQueryPlanProto(@Nonnull final PlanSerializationC
     @Nonnull
     public static RecordQueryStreamingAggregationPlan fromProto(@Nonnull final PlanSerializationContext serializationContext,
                                                                 @Nonnull final PRecordQueryStreamingAggregationPlan recordQueryStreamingAggregationPlanProto) {
-        return new RecordQueryStreamingAggregationPlan(Quantifier.Physical.fromProto(serializationContext, Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getInner())),
-                PlanSerialization.getFieldOrNull(recordQueryStreamingAggregationPlanProto,
-                        PRecordQueryStreamingAggregationPlan::hasGroupingKeyValue,
-                        m -> Value.fromValueProto(serializationContext, m.getGroupingKeyValue())),
-                (AggregateValue)Value.fromValueProto(serializationContext, Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getAggregateValue())),
-                CorrelationIdentifier.of(Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getGroupingKeyAlias())),
-                CorrelationIdentifier.of(Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getAggregateAlias())),
-                Value.fromValueProto(serializationContext, Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getCompleteResultValue())),
-                recordQueryStreamingAggregationPlanProto.hasIsCreateDefaultOnEmpty() ? recordQueryStreamingAggregationPlanProto.getIsCreateDefaultOnEmpty() : true);
+        // Note: it is important for proper deserialization (at least of things that interact with the serializationContext's cache of
+        // referenced values and plans) that we deserialize the values in the same order as they are serialized, or we may
+        // not
+        final Quantifier.Physical inner = Quantifier.Physical.fromProto(serializationContext, Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getInner()));
+        final AggregateValue aggregateValue =  (AggregateValue) Value.fromValueProto(serializationContext, Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getAggregateValue()));
+        @Nullable final Value groupingKeyValue = PlanSerialization.getFieldOrNull(recordQueryStreamingAggregationPlanProto, PRecordQueryStreamingAggregationPlan::hasGroupingKeyValue,
+                m -> Value.fromValueProto(serializationContext, m.getGroupingKeyValue()));
+        final CorrelationIdentifier groupingKeyAlias = CorrelationIdentifier.of(Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getGroupingKeyAlias()));
+        final CorrelationIdentifier aggregateAlias = CorrelationIdentifier.of(Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getAggregateAlias()));
+        final Value completeResultValue = Value.fromValueProto(serializationContext, Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getCompleteResultValue()));
+        final boolean isCreateDefaultOnEmpty = recordQueryStreamingAggregationPlanProto.hasIsCreateDefaultOnEmpty() ? recordQueryStreamingAggregationPlanProto.getIsCreateDefaultOnEmpty() : true;
+        return new RecordQueryStreamingAggregationPlan(inner, groupingKeyValue, aggregateValue, groupingKeyAlias, aggregateAlias, completeResultValue, isCreateDefaultOnEmpty);
     }
 
     @Nonnull

fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/query/FDBStreamAggregationTest.java

@@ -334,7 +334,8 @@ private void populateDB(final int numRecords) throws Exception {
     }
 
     @Nonnull
-    private RecordCursor<QueryResult> executePlan(final RecordQueryPlan plan, final int rowLimit, final byte[] continuation) {
+    private RecordCursor<QueryResult> executePlan(final RecordQueryPlan originalPlan, final int rowLimit, final byte[] continuation) {
+        final RecordQueryPlan plan = verifySerialization(originalPlan);
         final var types = plan.getDynamicTypes();
         final var typeRepository = TypeRepository.newBuilder().addAllTypes(types).build();
         ExecuteProperties executeProperties = ExecuteProperties.SERIAL_EXECUTE;

scripts/YAML-SQL.xml

@@ -51,8 +51,9 @@
     <keywords3 keywords="false;null;true;ordered;randomized;parallelized;test;block;single_repetition_ordered;
     single_repetition_randomized;single_repetition_parallelized;multi_repetition_ordered;multi_repetition_randomized;
     multi_repetition_parallelized" />
-    <keywords4 keywords="connect:;query:;load schema template:;set schema state:;result:;unorderedresult:;explain:;
-    explaincontains:;count:;error:;planhash:;setup:;schema_template:;supported_version:;test_block:;options:;tests:;mode:;repetition:;seed:;
+    <keywords4 keywords="connect:;query:;load schema template:;set schema state:;result:;unorderedResult:;explain:;
+    explainContains:;count:;error:;planHash:;setup:;schema_template:;supported_version:;test_block:;options:;tests:;
+    maxRows:;initialVersionAtLeast:;initialVersionLessThan:;;mode:;repetition:;seed:;
     check_cache:;connection_lifecycle:;steps:;preset:;statement_type:;!r;!in;!a;" />
   </highlighting>
   <extensionMap>

yaml-tests/src/main/java/com/apple/foundationdb/relational/yamltests/CustomYamlConstructor.java

@@ -75,6 +75,8 @@ public CustomYamlConstructor(LoaderOptions loaderOptions) {
         requireLineNumber.add(QueryConfig.QUERY_CONFIG_PLAN_HASH);
         requireLineNumber.add(QueryConfig.QUERY_CONFIG_MAX_ROWS);
         requireLineNumber.add(QueryConfig.QUERY_CONFIG_SUPPORTED_VERSION);
+        requireLineNumber.add(QueryConfig.QUERY_CONFIG_INITIAL_VERSION_LESS_THAN);
+        requireLineNumber.add(QueryConfig.QUERY_CONFIG_INITIAL_VERSION_AT_LEAST);
     }
 
     @Override

yaml-tests/src/main/java/com/apple/foundationdb/relational/yamltests/MultiServerConnectionFactory.java

@@ -25,6 +25,7 @@
 import com.apple.foundationdb.relational.api.metrics.MetricCollector;
 import com.apple.foundationdb.relational.recordlayer.EmbeddedRelationalConnection;
 import com.apple.foundationdb.relational.util.Assert;
+import com.apple.foundationdb.relational.yamltests.server.SemanticVersion;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.junit.jupiter.api.Assertions;
@@ -49,7 +50,7 @@
 public class MultiServerConnectionFactory implements YamlConnectionFactory {
     // The fixed index of the default connection
     public static final int DEFAULT_CONNECTION = 0;
-    private final Set<String> versionsUnderTest;
+    private final Set<SemanticVersion> versionsUnderTest;
 
     /**
      * Server selection policy.
@@ -96,7 +97,7 @@ public YamlConnection getNewConnection(@Nonnull URI connectPath) throws SQLExcep
     }
 
     @Override
-    public Set<String> getVersionsUnderTest() {
+    public Set<SemanticVersion> getVersionsUnderTest() {
         return versionsUnderTest;
     }
 
@@ -148,7 +149,7 @@ public static class MultiServerConnection implements YamlConnection {
         @Nonnull
         private final List<YamlConnection> underlyingConnections;
         @Nonnull
-        private final List<String> versions;
+        private final List<SemanticVersion> versions;
 
         public MultiServerConnection(@Nonnull ConnectionSelectionPolicy connectionSelectionPolicy,
                                      final int initialConnecion,
@@ -188,10 +189,16 @@ public EmbeddedRelationalConnection tryGetEmbedded() {
 
         @Nonnull
         @Override
-        public List<String> getVersions() {
+        public List<SemanticVersion> getVersions() {
             return this.versions;
         }
 
+        @Nonnull
+        @Override
+        public SemanticVersion getInitialVersion() {
+            return versions.get(0);
+        }
+
         @Override
         public void close() throws SQLException {
             logger.info("Sending operation {} to all connections", "close");
@@ -228,16 +235,16 @@ private YamlConnection getCurrentConnection(boolean advance, String op) {
             throw new IllegalStateException("Unsupported selection policy " + connectionSelectionPolicy);
         }
 
-        private static List<String> createVersionsList(final int initialConnecion,
-                                                       final List<YamlConnection> relationalConnections) {
-            List<String> versions = new ArrayList<>();
-            for (int i = initialConnecion; i < relationalConnections.size(); i++) {
-                final List<String> underlying = relationalConnections.get(i).getVersions();
+        private static List<SemanticVersion> createVersionsList(final int initialConnection,
+                                                                final List<YamlConnection> relationalConnections) {
+            List<SemanticVersion> versions = new ArrayList<>();
+            for (int i = initialConnection; i < relationalConnections.size(); i++) {
+                final List<SemanticVersion> underlying = relationalConnections.get(i).getVersions();
                 Assert.thatUnchecked(underlying.size() == 1, "Part of multi server config has more than one version");
                 versions.add(underlying.get(0));
             }
-            for (int i = 0; i < initialConnecion; i++) {
-                final List<String> underlying = relationalConnections.get(i).getVersions();
+            for (int i = 0; i < initialConnection; i++) {
+                final List<SemanticVersion> underlying = relationalConnections.get(i).getVersions();
                 Assert.thatUnchecked(underlying.size() == 1, "Part of multi server config has more than one version");
                 versions.add(underlying.get(0));
             }

yaml-tests/src/main/java/com/apple/foundationdb/relational/yamltests/SimpleYamlConnection.java

@@ -25,6 +25,7 @@
 import com.apple.foundationdb.relational.api.RelationalStatement;
 import com.apple.foundationdb.relational.api.metrics.MetricCollector;
 import com.apple.foundationdb.relational.recordlayer.EmbeddedRelationalConnection;
+import com.apple.foundationdb.relational.yamltests.server.SemanticVersion;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -39,9 +40,9 @@ public class SimpleYamlConnection implements YamlConnection {
     @Nonnull
     private final RelationalConnection underlying;
     @Nonnull
-    private final List<String> versions;
+    private final List<SemanticVersion> versions;
 
-    public SimpleYamlConnection(@Nonnull Connection connection, @Nonnull String version) throws SQLException {
+    public SimpleYamlConnection(@Nonnull Connection connection, @Nonnull SemanticVersion version) throws SQLException {
         underlying = connection.unwrap(RelationalConnection.class);
         this.versions = List.of(version);
     }
@@ -84,7 +85,13 @@ public EmbeddedRelationalConnection tryGetEmbedded() {
 
     @Nonnull
     @Override
-    public List<String> getVersions() {
+    public List<SemanticVersion> getVersions() {
         return versions;
     }
+
+    @Nonnull
+    @Override
+    public SemanticVersion getInitialVersion() {
+        return versions.get(0);
+    }
 }

yaml-tests/src/main/java/com/apple/foundationdb/relational/yamltests/YamlConnection.java

@@ -25,6 +25,7 @@
 import com.apple.foundationdb.relational.api.RelationalStatement;
 import com.apple.foundationdb.relational.api.metrics.MetricCollector;
 import com.apple.foundationdb.relational.recordlayer.EmbeddedRelationalConnection;
+import com.apple.foundationdb.relational.yamltests.server.SemanticVersion;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -35,11 +36,6 @@
  * A wrapper around {@link java.sql.Connection} to support yaml tests.
  */
 public interface YamlConnection extends AutoCloseable {
-    /**
-     * String for representing the current version of the code.
-     */
-    String CURRENT_VERSION = "!currentVersion";
-
     /**
      * Close this connection.
      * @throws SQLException if there was an issue closing the underlying connection(s).
@@ -91,5 +87,18 @@ public interface YamlConnection extends AutoCloseable {
      * @return the ordered list of versions
      */
     @Nonnull
-    List<String> getVersions();
+    List<SemanticVersion> getVersions();
+
+    /**
+     * Return the initial version returned by this connection. If this connection
+     * wraps multiple versions, it may return different underlying connections
+     * with every call to {@link #createStatement()}. This returns the version
+     * associated with the first such call, which can impact the set of results
+     * that we expect to return as that is also the connection that should be
+     * used for query planning.
+     *
+     * @return the first version that an underlying connection will represent
+     */
+    @Nonnull
+    SemanticVersion getInitialVersion();
 }

yaml-tests/src/main/java/com/apple/foundationdb/relational/yamltests/YamlConnectionFactory.java

@@ -21,6 +21,7 @@
 package com.apple.foundationdb.relational.yamltests;
 
 import com.apple.foundationdb.relational.api.RelationalConnection;
+import com.apple.foundationdb.relational.yamltests.server.SemanticVersion;
 
 import javax.annotation.Nonnull;
 import java.net.URI;
@@ -53,7 +54,7 @@ public interface YamlConnectionFactory {
      * @return A set of versions that we are testing against, or an empty set if just testing against the current
      * version
      */
-    Set<String> getVersionsUnderTest();
+    Set<SemanticVersion> getVersionsUnderTest();
 
     /**
      * Whether the connection supports multiple servers.

yaml-tests/src/main/java/com/apple/foundationdb/relational/yamltests/YamlExecutionContext.java

@@ -190,12 +190,7 @@ public Optional<Integer> getNightlyRepetition() {
     }
 
     public int getNumThreads() {
-        var numThreads = 1;
-        if (System.getProperties().stringPropertyNames().contains(YamlRunner.TEST_MAX_THREADS)) {
-            numThreads = Integer.parseInt(System.getProperty(YamlRunner.TEST_MAX_THREADS));
-            Assert.thatUnchecked(numThreads > 0, "Invalid number of threads provided in the YamlExecutionContext");
-        }
-        return numThreads;
+        return Runtime.getRuntime().availableProcessors() / 2;
     }
 
     boolean isDirty() {

yaml-tests/src/main/java/com/apple/foundationdb/relational/yamltests/YamlRunner.java

@@ -51,10 +51,9 @@ public final class YamlRunner {
 
     private static final Logger logger = LogManager.getLogger(YamlRunner.class);
 
-    static final String TEST_NIGHTLY = "yaml_testing_nightly";
-    static final String TEST_SEED = "yaml_testing_seed";
-    static final String TEST_NIGHTLY_REPETITION = "yaml_testing_nightly_repetition";
-    static final String TEST_MAX_THREADS = "yaml_testing_max_threads";
+    static final String TEST_NIGHTLY = "tests.nightly";
+    static final String TEST_SEED = "tests.yaml.seed";
+    static final String TEST_NIGHTLY_REPETITION = "tests.yaml.iterations";
 
     @Nonnull
     private final String resourcePath;

yaml-tests/src/main/java/com/apple/foundationdb/relational/yamltests/block/FileOptions.java

@@ -23,6 +23,7 @@
 import com.apple.foundationdb.relational.yamltests.CustomYamlConstructor;
 import com.apple.foundationdb.relational.yamltests.Matchers;
 import com.apple.foundationdb.relational.yamltests.YamlExecutionContext;
+import com.apple.foundationdb.relational.yamltests.server.SemanticVersion;
 import com.apple.foundationdb.relational.yamltests.server.SupportedVersionCheck;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -55,8 +56,7 @@ public class FileOptions {
 
     public static Block parse(int lineNumber, Object document, YamlExecutionContext executionContext) {
         final Map<?, ?> options = CustomYamlConstructor.LinedObject.unlineKeys(Matchers.map(document, OPTIONS));
-        Object rawVersion = options.get(SUPPORTED_VERSION_OPTION);
-        final SupportedVersionCheck check = SupportedVersionCheck.parse(rawVersion, executionContext);
+        final SupportedVersionCheck check = SupportedVersionCheck.parseOptions(options, executionContext);
         if (!check.isSupported()) {
             // IntelliJ, at least, doesn't display the reason, so log it
             if (logger.isInfoEnabled()) {
@@ -67,15 +67,26 @@ public static Block parse(int lineNumber, Object document, YamlExecutionContext
         return new NoOpBlock(lineNumber);
     }
 
+    public static SemanticVersion parseVersion(Object rawVersion) {
+        if (rawVersion instanceof CurrentVersion) {
+            return SemanticVersion.current();
+        } else if (rawVersion instanceof String) {
+            return SemanticVersion.parse((String) rawVersion);
+        } else {
+            throw new IllegalArgumentException("Unable to determine semantic version from object: " + rawVersion);
+        }
+    }
+
     public static final class CurrentVersion {
         public static final CurrentVersion INSTANCE = new CurrentVersion();
+        public static final String TEXT = SemanticVersion.SemanticVersionType.CURRENT.getText();
 
         private CurrentVersion() {
         }
 
         @Override
         public String toString() {
-            return "!current_version";
+            return TEXT;
         }
     }
 }