Fix/temporaljoin (#1241)

* add test case for temporal join against kafka
* reduce time delay
* update test runner to add a configurable delay. Update temporal join test to use it.
* set config parameters before compilation so they become part of the compiled plan.
* update documentation

Author: mbroecheler

documentation/docs/configuration.md

@@ -37,6 +37,10 @@ Connectors that connect flink to other engines and external systems can be confi
 
 Environment variables that start with the `sqrl` prefix are templated variables that the DataSQRL compiler instantiates. For example: `${sqrl:table-name}` provides the table name for a connector that writes to a table.
 
+Flink configuration can be specified in the `config` section. See [Flink configuration](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/table/config/) for details.
+These configuration options are applied when compiling the Flink plan and when executing Flink while running or testing via the [DataSQRL command](compiler). 
+
+
 ```json
 {
   "engines" : {
@@ -60,25 +64,15 @@ Environment variables that start with the `sqrl` prefix are templated variables
           "properties.auto.offset.reset" : "earliest",
           "topic" : "${sqrl:original-table-name}"
         }
+      },
+      "config" : {
+        "table.exec.source.idle-timeout": "5000 ms"
       }
     }
   }
 }
 ```
 
-Flink runtime configuration can be specified in the [`values` configuration](#values) section:
-
-```json
-{
-  "values" : {
-    "flink-config": {
-      "table.exec.source.idle-timeout": "100 ms"
-    }
-  }
-}
-```
-The configuration options are set on the Flink runtime when running or testing via the [DataSQRL command](compiler).
-
 ### Postgres
 
 Postgres is the default database engine.
@@ -256,36 +250,29 @@ Testing related configuration is found in the `test-runner` section.
 ```json
 {
   "test-runner": {
-    "delay-sec": 30
+    "delay-sec": 30,
+    "mutation-delay": 1
   }
 }
 ```
 
-* `delay-sec`: The number of seconds to wait between starting the processing of data and snapshotting the data.
-
+* `delay-sec`: The number of seconds to wait between starting the processing of data and snapshotting the data. Defaults to 30.
+* `required-checkpoints`: The number of checkpoints that need to be completed before the test queries are executed. Defaults to 0. If this is configured to a positive value, `delay-sec` must be set to -1.
+* `mutation-delay`: The number of seconds to wait between execution mutation queries. Defaults to 0.
 
 
 ## Values
 
-The `values` section of the configuration allows you to specify configuration values that are passed through to engines they pertain to.
-
-The default deployment profiles supports a `flink-config` section to allow injecting additional flink runtime configuration. You can use this section of the configuration to specify any [Flink configuration option](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/table/config/).
+The `values` section of the configuration allows you to specify configuration values that is applied at runtime when running or testing SQRL scripts with the [DataSQRL command](compiler).
 
 ```json
 {
  "values" : {
-    "flink-config" : {
-      "taskmanager.memory.network.max": "800m",
-      "execution.checkpointing.mode" : "EXACTLY_ONCE",
-      "execution.checkpointing.interval" : "1000ms"
-    },
     "create-topics": ["mytopic"]
    }
 }
 ```
 
-For Flink, the `values` configuration settings take precedence over identical configuration settings in the compiled physical plans.
-
 For the log engine, the `create-topics` option allows you to specify topics to create in the cluster prior to starting the pipeline. This is useful for testing.
 
 ## Environmental Variables

sqrl-planner/src/main/java/com/datasqrl/config/PackageJson.java

@@ -39,8 +39,6 @@ public interface PackageJson {
 
   ScriptConfig getScriptConfig();
 
-  Map<String, Object> toMap();
-
   CompilerConfig getCompilerConfig();
 
   PackageConfiguration getPackageConfig();
@@ -117,6 +115,8 @@ interface EngineConfig {
     String getSetting(String key, Optional<String> defaultValue);
 
     ConnectorsConfig getConnectors();
+
+    Map<String, Object> getConfig();
   }
 
   @AllArgsConstructor
@@ -136,6 +136,11 @@ public String getSetting(String key, Optional<String> defaultValue) {
     public ConnectorsConfig getConnectors() {
       return null;
     }
+
+    @Override
+    public Map<String, Object> getConfig() {
+      return Map.of();
+    }
   }
 
   interface DependenciesConfig {

sqrl-planner/src/main/java/com/datasqrl/engine/stream/flink/FlinkStreamEngine.java

@@ -28,7 +28,9 @@
 import com.google.inject.Inject;
 import java.io.IOException;
 import java.util.EnumSet;
+import java.util.HashMap;
 import java.util.Locale;
+import java.util.Map;
 import java.util.Optional;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
@@ -40,12 +42,12 @@ public class FlinkStreamEngine extends ExecutionEngine.Base implements StreamEng
 
   public static final EnumSet<EngineFeature> FLINK_CAPABILITIES = STANDARD_STREAM;
 
-  @Getter private final EngineConfig config;
+  @Getter private final EngineConfig engineConfig;
 
   @Inject
   public FlinkStreamEngine(PackageJson json) {
     super(FlinkEngineFactory.ENGINE_NAME, EngineType.STREAMS, FLINK_CAPABILITIES);
-    this.config =
+    this.engineConfig =
         json.getEngines()
             .getEngineConfig(FlinkEngineFactory.ENGINE_NAME)
             .orElseGet(() -> new EmptyEngineConfig(FlinkEngineFactory.ENGINE_NAME));
@@ -57,8 +59,14 @@ public void close() throws IOException {}
   @Override
   public ExecutionMode getExecutionMode() {
     return ExecutionMode.valueOf(
-        config
+        engineConfig
             .getSetting(MODE_KEY, Optional.of(ExecutionMode.STREAMING.name()))
             .toUpperCase(Locale.ENGLISH));
   }
+
+  public Map<String, String> getBaseConfiguration() {
+    Map<String, String> configMap = new HashMap<>();
+    engineConfig.getConfig().forEach((key, value) -> configMap.put(key, String.valueOf(value)));
+    return configMap;
+  }
 }

sqrl-planner/src/main/java/com/datasqrl/planner/Sqrl2FlinkSQLTranslator.java

@@ -58,7 +58,6 @@
 import java.nio.file.Files;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -182,7 +181,7 @@ public Sqrl2FlinkSQLTranslator(
     // Create a UDF class loader and configure
     ClassLoader udfClassLoader =
         new URLClassLoader(jarUrls.toArray(new URL[0]), getClass().getClassLoader());
-    Map<String, String> config = new HashMap<>();
+    Map<String, String> config = flink.getBaseConfiguration();
     config.put(
         "pipeline.classpaths",
         jarUrls.stream().map(URL::toString).collect(Collectors.joining(",")));
@@ -215,7 +214,7 @@ public Sqrl2FlinkSQLTranslator(
     var calciteConfigBuilder = new CalciteConfigBuilder();
     calciteConfigBuilder.addSqlOperatorTable(sqrlFunctionCatalog.getOperatorTable());
     //    setOptimizerRules(calciteConfigBuilder,tEnv.getConfig()); TODO: fix, so we have more
-    // effective subgraph identification
+    // effective subgraph identification by not pushing down projections
 
     this.tEnv.getConfig().setPlannerConfig(calciteConfigBuilder.build());
     this.catalogManager = tEnv.getCatalogManager();

sqrl-testing/sqrl-integration-tests/src/test/java/com/datasqrl/FullUsecasesIT.java

@@ -96,7 +96,11 @@ class ScriptCriteria {
           new ScriptCriteria("postgres-log-disabled.sqrl", "test"),
           new ScriptCriteria("postgres-log-disabled.sqrl", "run"),
           new ScriptCriteria("connectors.sqrl", "test"), // should not be executed
-          new ScriptCriteria("flink_kafka.sqrl", "run") // does not expose an API
+          new ScriptCriteria("flink_kafka.sqrl", "run"), // does not expose an API
+          new ScriptCriteria(
+              "temporal-join.sqrl",
+              "run") // TODO: only 'run' when there are no tests (i.e. snapshot dir) - there is no
+          // benefit to also running, it's wasteful
           );
 
   static final Path PROJECT_ROOT = Path.of(System.getProperty("user.dir"));
@@ -386,7 +390,7 @@ public void printUseCaseNumbers(UseCaseTestParameter param) {
   @MethodSource("useCaseProvider")
   @Disabled
   public void runTestCaseByName(UseCaseTestParameter param) {
-    if (param.sqrlFileName.equals("banking-batch.sqrl") && param.goal.equals("test")) {
+    if (param.sqrlFileName.equals("temporal-join.sqrl") && param.goal.equals("test")) {
       testUseCase(param);
     } else {
       assumeFalse(true);