[FEATURE] Add PPL asynchronous query lifecycle core

common/src/main/java/org/opensearch/sql/common/setting/Settings.java

@@ -38,6 +38,18 @@ public enum Key {
     PPL_JOIN_SUBSEARCH_MAXOUT("plugins.ppl.join.subsearch_maxout"),
     PPL_REST_ALLOWED_ENDPOINTS("plugins.ppl.rest.allowed_endpoints"),
 
+    /** Maximum number of asynchronous PPL queries running concurrently on one node. */
+    PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES("plugins.ppl.async.node_concurrent_running_queries"),
+
+    /** Maximum number of asynchronous PPL jobs retained on one node. */
+    PPL_ASYNC_MAX_RETAINED_JOBS("plugins.ppl.async.max_retained_jobs"),
+
+    /** Maximum accepted submit wait-for-completion timeout. */
+    PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT("plugins.ppl.async.max_wait_for_completion_timeout"),
+
+    /** Maximum accepted asynchronous PPL job lease. */
+    PPL_ASYNC_MAX_KEEP_ALIVE("plugins.ppl.async.max_keep_alive"),
+
     /** Enable Calcite as execution engine */
     CALCITE_ENGINE_ENABLED("plugins.calcite.enabled"),
     CALCITE_FALLBACK_ALLOWED("plugins.calcite.fallback.allowed"),

core/src/main/java/org/opensearch/sql/executor/AsyncQueryExecution.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.executor;
+
+import java.util.Optional;
+import java.util.concurrent.CompletionStage;
+import org.opensearch.sql.executor.ExecutionEngine.QueryResponse;
+
+/**
+ * Lifecycle-facing handle for one asynchronous query execution.
+ *
+ * <p>The execution module owns result production and execution-specific resources. The lifecycle
+ * module owns this handle after submission and uses it to read the current result, observe terminal
+ * completion, and release those resources.
+ *
+ * <p>On successful completion, the authoritative final result must be visible through {@link
+ * #currentResult()} before {@link #completion()} completes normally. Implementations must make
+ * {@link #close()} idempotent and safe to call concurrently with {@link #currentResult()}.
+ */
+public interface AsyncQueryExecution extends AutoCloseable {
+
+  /** Returns the complete result currently visible, or empty before a result is available. */
+  Optional<QueryResponse> currentResult();
+
+  /** Completes normally on query success and exceptionally on query failure. */
+  CompletionStage<Void> completion();
+
+  /** Releases execution-owned result resources. */
+  @Override
+  void close();
+}

opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java

@@ -36,6 +36,8 @@
 /** Setting implementation on OpenSearch. */
 @Log4j2
 public class OpenSearchSettings extends Settings {
+  private static final TimeValue MAX_PPL_ASYNC_KEEP_ALIVE = TimeValue.timeValueHours(24);
+
   /** Default settings. */
   private final Map<Settings.Key, Setting<?>> defaultSettings;
 
@@ -86,6 +88,42 @@ public class OpenSearchSettings extends Settings {
           Setting.Property.NodeScope,
           Setting.Property.Dynamic);
 
+  /** Maximum number of asynchronous PPL queries running concurrently on one node. */
+  public static final Setting<Integer> PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES_SETTING =
+      Setting.intSetting(
+          Key.PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES.getKeyValue(),
+          20,
+          1,
+          Setting.Property.NodeScope,
+          Setting.Property.Dynamic);
+
+  /** Maximum number of asynchronous PPL jobs retained on one node. */
+  public static final Setting<Integer> PPL_ASYNC_MAX_RETAINED_JOBS_SETTING =
+      Setting.intSetting(
+          Key.PPL_ASYNC_MAX_RETAINED_JOBS.getKeyValue(),
+          100,
+          1,
+          Setting.Property.NodeScope,
+          Setting.Property.Dynamic);
+
+  /** Maximum accepted submit wait-for-completion timeout. */
+  public static final Setting<TimeValue> PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT_SETTING =
+      Setting.positiveTimeSetting(
+          Key.PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT.getKeyValue(),
+          TimeValue.timeValueSeconds(60),
+          Setting.Property.NodeScope,
+          Setting.Property.Dynamic);
+
+  /** Configurable asynchronous PPL job lease limit, capped at 24 hours. */
+  public static final Setting<TimeValue> PPL_ASYNC_MAX_KEEP_ALIVE_SETTING =
+      Setting.timeSetting(
+          Key.PPL_ASYNC_MAX_KEEP_ALIVE.getKeyValue(),
+          MAX_PPL_ASYNC_KEEP_ALIVE,
+          TimeValue.ZERO,
+          MAX_PPL_ASYNC_KEEP_ALIVE,
+          Setting.Property.NodeScope,
+          Setting.Property.Dynamic);
+
   public static final Setting<?> PPL_SYNTAX_LEGACY_PREFERRED_SETTING =
       Setting.boolSetting(
           Key.PPL_SYNTAX_LEGACY_PREFERRED.getKeyValue(),
@@ -444,6 +482,30 @@ public OpenSearchSettings(ClusterSettings clusterSettings) {
         Key.PPL_QUERY_TIMEOUT,
         PPL_QUERY_TIMEOUT_SETTING,
         new Updater(Key.PPL_QUERY_TIMEOUT));
+    register(
+        settingBuilder,
+        clusterSettings,
+        Key.PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES,
+        PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES_SETTING,
+        new Updater(Key.PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES));
+    register(
+        settingBuilder,
+        clusterSettings,
+        Key.PPL_ASYNC_MAX_RETAINED_JOBS,
+        PPL_ASYNC_MAX_RETAINED_JOBS_SETTING,
+        new Updater(Key.PPL_ASYNC_MAX_RETAINED_JOBS));
+    register(
+        settingBuilder,
+        clusterSettings,
+        Key.PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT,
+        PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT_SETTING,
+        new Updater(Key.PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT));
+    register(
+        settingBuilder,
+        clusterSettings,
+        Key.PPL_ASYNC_MAX_KEEP_ALIVE,
+        PPL_ASYNC_MAX_KEEP_ALIVE_SETTING,
+        new Updater(Key.PPL_ASYNC_MAX_KEEP_ALIVE));
     register(
         settingBuilder,
         clusterSettings,
@@ -767,6 +829,10 @@ public static List<Setting<?>> pluginSettings() {
         .add(DESERIALIZATION_MAX_BYTES_SETTING)
         .add(PPL_ENABLED_SETTING)
         .add(PPL_QUERY_TIMEOUT_SETTING)
+        .add(PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES_SETTING)
+        .add(PPL_ASYNC_MAX_RETAINED_JOBS_SETTING)
+        .add(PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT_SETTING)
+        .add(PPL_ASYNC_MAX_KEEP_ALIVE_SETTING)
         .add(PPL_SYNTAX_LEGACY_PREFERRED_SETTING)
         .add(CALCITE_ENGINE_ENABLED_SETTING)
         .add(CALCITE_FALLBACK_ALLOWED_SETTING)

opensearch/src/test/java/org/opensearch/sql/opensearch/setting/OpenSearchSettingsTest.java

@@ -9,6 +9,7 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.AdditionalMatchers.not;
 import static org.mockito.AdditionalMatchers.or;
@@ -28,6 +29,7 @@
 import org.opensearch.cluster.ClusterName;
 import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Setting;
+import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.common.unit.ByteSizeValue;
 import org.opensearch.sql.common.setting.Settings;
 import org.opensearch.sql.utils.DeserializationFilterUtil;
@@ -168,6 +170,50 @@ void deserializationStructuralLimitSettingsAreDynamicAndRegistered() {
     assertTrue(pluginSettings.contains(OpenSearchSettings.DESERIALIZATION_MAX_BYTES_SETTING));
   }
 
+  @Test
+  void pplAsyncSettingsHaveExpectedDefaultsAndAreDynamic() {
+    org.opensearch.common.settings.Settings empty = org.opensearch.common.settings.Settings.EMPTY;
+
+    assertEquals(
+        20, OpenSearchSettings.PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES_SETTING.get(empty));
+    assertEquals(100, OpenSearchSettings.PPL_ASYNC_MAX_RETAINED_JOBS_SETTING.get(empty));
+    assertEquals(
+        TimeValue.timeValueSeconds(60),
+        OpenSearchSettings.PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT_SETTING.get(empty));
+    assertEquals(
+        TimeValue.timeValueHours(24),
+        OpenSearchSettings.PPL_ASYNC_MAX_KEEP_ALIVE_SETTING.get(empty));
+
+    List<Setting<?>> pluginSettings = OpenSearchSettings.pluginSettings();
+    assertTrue(
+        pluginSettings.contains(
+            OpenSearchSettings.PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES_SETTING));
+    assertTrue(pluginSettings.contains(OpenSearchSettings.PPL_ASYNC_MAX_RETAINED_JOBS_SETTING));
+    assertTrue(
+        pluginSettings.contains(
+            OpenSearchSettings.PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT_SETTING));
+    assertTrue(pluginSettings.contains(OpenSearchSettings.PPL_ASYNC_MAX_KEEP_ALIVE_SETTING));
+    assertTrue(OpenSearchSettings.PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES_SETTING.isDynamic());
+    assertTrue(OpenSearchSettings.PPL_ASYNC_MAX_RETAINED_JOBS_SETTING.isDynamic());
+    assertTrue(OpenSearchSettings.PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT_SETTING.isDynamic());
+    assertTrue(OpenSearchSettings.PPL_ASYNC_MAX_KEEP_ALIVE_SETTING.isDynamic());
+  }
+
+  @Test
+  void pplAsyncMaxKeepAliveIsCappedAt24Hours() {
+    String key = Settings.Key.PPL_ASYNC_MAX_KEEP_ALIVE.getKeyValue();
+
+    assertEquals(
+        TimeValue.timeValueHours(24),
+        OpenSearchSettings.PPL_ASYNC_MAX_KEEP_ALIVE_SETTING.get(
+            org.opensearch.common.settings.Settings.builder().put(key, "24h").build()));
+    assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            OpenSearchSettings.PPL_ASYNC_MAX_KEEP_ALIVE_SETTING.get(
+                org.opensearch.common.settings.Settings.builder().put(key, "25h").build()));
+  }
+
   @Test
   void getSparkExecutionEngineConfigSetting() {
     // Default is empty string

plugin/build.gradle

@@ -168,6 +168,7 @@ dependencies {
     api project(':datasources')
     api project(':async-query')
     api project(':direct-query')
+    implementation "org.opensearch:common-utils:${opensearch_build}"
 
     testImplementation group: 'net.bytebuddy', name: 'byte-buddy-agent', version: '1.15.11'
     testImplementation group: 'org.hamcrest', name: 'hamcrest-library', version: "${hamcrest_version}"

plugin/src/main/java/org/opensearch/sql/plugin/PPLQueryErrorHandler.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.plugin;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.OpenSearchException;
+import org.opensearch.core.rest.RestStatus;
+import org.opensearch.index.IndexNotFoundException;
+import org.opensearch.sql.common.antlr.SyntaxCheckException;
+import org.opensearch.sql.common.error.ErrorReport;
+import org.opensearch.sql.datasources.exceptions.DataSourceClientException;
+import org.opensearch.sql.exception.QueryEngineException;
+import org.opensearch.sql.legacy.metrics.MetricName;
+import org.opensearch.sql.legacy.metrics.Metrics;
+
+/** Classifies PPL failures and records the corresponding customer or system error metric. */
+public final class PPLQueryErrorHandler {
+  private static final Logger LOG = LogManager.getLogger(PPLQueryErrorHandler.class);
+
+  private PPLQueryErrorHandler() {}
+
+  /**
+   * Records a PPL failure and returns the HTTP status associated with it.
+   *
+   * @param exception query failure
+   * @return client or system error status
+   */
+  public static RestStatus recordFailure(Exception exception) {
+    int code = rawStatusCode(exception);
+    if (400 <= code && code < 500) {
+      increment(MetricName.PPL_FAILED_REQ_COUNT_CUS);
+    } else if (500 <= code && code < 600) {
+      increment(MetricName.PPL_FAILED_REQ_COUNT_SYS);
+    } else {
+      LOG.warn(
+          "Got an exception returning non-error status {}", RestStatus.fromCode(code), exception);
+    }
+    return RestStatus.fromCode(code);
+  }
+
+  private static int rawStatusCode(Exception exception) {
+    if (exception instanceof ErrorReport errorReport) {
+      return rawStatusCode(errorReport.getCause());
+    }
+    if (exception instanceof OpenSearchException openSearchException) {
+      return openSearchException.status().getStatus();
+    }
+    return isClientError(exception) ? 400 : 500;
+  }
+
+  private static boolean isClientError(Exception exception) {
+    return exception instanceof IllegalArgumentException
+        || exception instanceof IndexNotFoundException
+        || exception instanceof QueryEngineException
+        || exception instanceof SyntaxCheckException
+        || exception instanceof DataSourceClientException
+        || exception instanceof IllegalAccessException;
+  }
+
+  private static void increment(MetricName metricName) {
+    Metrics.getInstance().getNumericalMetric(metricName).increment();
+  }
+}