feat: PPL OpenTelemetry tracing integration

core/src/main/java/org/opensearch/sql/calcite/utils/CalciteToolsHelper.java

@@ -28,7 +28,6 @@
 package org.opensearch.sql.calcite.utils;
 
 import static java.util.Objects.requireNonNull;
-import static org.opensearch.sql.monitor.profile.MetricName.OPTIMIZE;
 
 import com.google.common.collect.ImmutableList;
 import java.lang.reflect.Type;
@@ -109,8 +108,9 @@
 import org.opensearch.sql.common.error.ErrorCode;
 import org.opensearch.sql.common.error.ErrorReport;
 import org.opensearch.sql.expression.function.PPLBuiltinOperators;
+import org.opensearch.sql.monitor.profile.MetricName;
 import org.opensearch.sql.monitor.profile.ProfileContext;
-import org.opensearch.sql.monitor.profile.ProfileMetric;
+import org.opensearch.sql.monitor.profile.ProfileScope;
 import org.opensearch.sql.monitor.profile.QueryProfiling;
 
 /**
@@ -514,37 +514,35 @@ private static void enrichErrorsForSpecialCases(ErrorReport.Builder report, SQLE
      * org.apache.calcite.tools.RelRunners#run(RelNode)}
      */
     public static PreparedStatement run(CalcitePlanContext context, RelNode rel) {
-      ProfileMetric optimizeTime = QueryProfiling.current().getOrCreateMetric(OPTIMIZE);
-      long startTime = System.nanoTime();
-      final RelShuttle shuttle =
-          new RelHomogeneousShuttle() {
-            @Override
-            public RelNode visit(TableScan scan) {
-              final RelOptTable table = scan.getTable();
-              if (scan instanceof LogicalTableScan
-                  && Bindables.BindableTableScan.canHandle(table)) {
-                // Always replace the LogicalTableScan with BindableTableScan
-                // because it's implementation does not require a "schema" as context.
-                return Bindables.BindableTableScan.create(scan.getCluster(), table);
+      try (ProfileScope optimizePhase = ProfileScope.open(MetricName.OPTIMIZE)) {
+        final RelShuttle shuttle =
+            new RelHomogeneousShuttle() {
+              @Override
+              public RelNode visit(TableScan scan) {
+                final RelOptTable table = scan.getTable();
+                if (scan instanceof LogicalTableScan
+                    && Bindables.BindableTableScan.canHandle(table)) {
+                  // Always replace the LogicalTableScan with BindableTableScan
+                  // because it's implementation does not require a "schema" as context.
+                  return Bindables.BindableTableScan.create(scan.getCluster(), table);
+                }
+                return super.visit(scan);
               }
-              return super.visit(scan);
-            }
-          };
-      rel = rel.accept(shuttle);
-      // the line we changed here
-      try (Connection connection = context.connection) {
-        final RelRunner runner = connection.unwrap(RelRunner.class);
-        PreparedStatement preparedStatement = runner.prepareStatement(rel);
-        optimizeTime.set(System.nanoTime() - startTime);
-        return preparedStatement;
-      } catch (SQLException e) {
-        // Detect if error is due to window functions in unsupported context (bins on time fields)
-        ErrorReport.Builder report =
-            ErrorReport.wrap(e)
-                .location("while compiling the optimized query plan for physical execution")
-                .code(ErrorCode.PLANNING_ERROR);
-        enrichErrorsForSpecialCases(report, e);
-        throw report.build();
+            };
+        rel = rel.accept(shuttle);
+
+        try (Connection connection = context.connection) {
+          final RelRunner runner = connection.unwrap(RelRunner.class);
+          return runner.prepareStatement(rel);
+        } catch (SQLException e) {
+          // Detect if error is due to window functions in unsupported context (bins on time fields)
+          ErrorReport.Builder report =
+              ErrorReport.wrap(e)
+                  .location("while compiling the optimized query plan for physical execution")
+                  .code(ErrorCode.PLANNING_ERROR);
+          enrichErrorsForSpecialCases(report, e);
+          throw report.build();
+        }
       }
     }
   }

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

@@ -61,8 +61,8 @@
 import org.opensearch.sql.exception.NonFallbackCalciteException;
 import org.opensearch.sql.executor.analyze.AnalyzeRecommendationBuilder;
 import org.opensearch.sql.monitor.profile.MetricName;
-import org.opensearch.sql.monitor.profile.ProfileContext;
 import org.opensearch.sql.monitor.profile.ProfileMetric;
+import org.opensearch.sql.monitor.profile.ProfileScope;
 import org.opensearch.sql.monitor.profile.QueryProfile;
 import org.opensearch.sql.monitor.profile.QueryProfiling;
 import org.opensearch.sql.planner.PlanContext;
@@ -219,38 +219,40 @@ public void executeWithCalcite(
     CalcitePlanContext.run(
         () -> {
           try {
-            ProfileContext profileContext =
-                QueryProfiling.activate(QueryContext.isProfileEnabled());
-            ProfileMetric analyzeMetric = profileContext.getOrCreateMetric(MetricName.ANALYZE);
-            long analyzeStart = System.nanoTime();
+            QueryProfiling.activate(QueryContext.isProfileEnabled());
             CalciteClassLoaderHelper.withCalciteClassLoader(
                 () -> {
-                  CalcitePlanContext context =
-                      CalcitePlanContext.create(
-                          buildFrameworkConfig(),
-                          SysLimit.fromSettings(settings),
-                          queryType,
-                          includeMetadata);
-
-                  context.setHighlightConfig(highlightConfig);
-
-                  // Wrap analyze with ANALYZING stage tracking
-                  RelNode relNode =
-                      StageErrorHandler.executeStage(
-                          QueryProcessingStage.ANALYZING,
-                          () -> analyze(plan, context),
-                          "while preparing and validating the query plan");
-
-                  // Wrap plan conversion with PLAN_CONVERSION stage tracking
-                  RelNode calcitePlan =
-                      StageErrorHandler.executeStage(
-                          QueryProcessingStage.PLAN_CONVERSION,
-                          () ->
-                              withCheckedArithmetic(
-                                  convertToCalcitePlan(relNode, context), context),
-                          "while converting the query to an executable plan");
-
-                  executeCalcitePlan(calcitePlan, context, listener, analyzeMetric, analyzeStart);
+                  CalcitePlanContext context;
+                  RelNode optimizedPlan;
+                  try (ProfileScope analyzePhase = ProfileScope.open(MetricName.ANALYZE)) {
+                    context =
+                        CalcitePlanContext.create(
+                            buildFrameworkConfig(),
+                            SysLimit.fromSettings(settings),
+                            queryType,
+                            includeMetadata);
+
+                    context.setHighlightConfig(highlightConfig);
+
+                    // Wrap analyze with ANALYZING stage tracking
+                    RelNode relNode =
+                        StageErrorHandler.executeStage(
+                            QueryProcessingStage.ANALYZING,
+                            () -> analyze(plan, context),
+                            "while preparing and validating the query plan");
+
+                    // Wrap plan conversion with PLAN_CONVERSION stage tracking
+                    RelNode calcitePlan =
+                        StageErrorHandler.executeStage(
+                            QueryProcessingStage.PLAN_CONVERSION,
+                            () ->
+                                withCheckedArithmetic(
+                                    convertToCalcitePlan(relNode, context), context),
+                            "while converting the query to an executable plan");
+
+                    optimizedPlan = CalciteToolsHelper.optimize(calcitePlan, context);
+                  }
+                  executeCalcitePlan(optimizedPlan, context, listener);
                 },
                 QueryService.class);
           } catch (Throwable t) {
@@ -266,17 +268,10 @@ public void executeWithCalcite(
   }
 
   private void executeCalcitePlan(
-      RelNode calcitePlan,
+      RelNode optimizedPlan,
       CalcitePlanContext context,
-      ResponseListener<ExecutionEngine.QueryResponse> listener,
-      ProfileMetric analyzeMetric,
-      long analyzeStart) {
+      ResponseListener<ExecutionEngine.QueryResponse> listener) {
     try {
-      // Optimize before dispatch so the dispatcher's ScriptDetector
-      // sees the post-optimization plan for accurate routing.
-      RelNode optimizedPlan = CalciteToolsHelper.optimize(calcitePlan, context);
-      analyzeMetric.set(System.nanoTime() - analyzeStart);
-
       // Wrap execution with EXECUTING stage tracking — dispatch via
       // ExecutionDispatcher which may route to a complex worker pool
       StageErrorHandler.executeStageVoid(
@@ -325,9 +320,13 @@ public void explainWithCalcite(
                   context.setHighlightConfig(highlightConfig);
                   context.run(
                       () -> {
-                        RelNode relNode = analyze(plan, context);
-                        RelNode calcitePlan =
-                            withCheckedArithmetic(convertToCalcitePlan(relNode, context), context);
+                        RelNode calcitePlan;
+                        try (ProfileScope analyzePhase = ProfileScope.open(MetricName.ANALYZE)) {
+                          RelNode relNode = analyze(plan, context);
+                          calcitePlan =
+                              withCheckedArithmetic(
+                                  convertToCalcitePlan(relNode, context), context);
+                        }
                         if (format != null) {
                           executionEngine.explain(calcitePlan, mode, format, context, listener);
                         } else {

core/src/main/java/org/opensearch/sql/monitor/profile/PhaseListener.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.monitor.profile;
+
+/**
+ * Observes query-phase boundaries — parse, analyze, optimize, compile, etc. — without pulling any
+ * tracing dependency into core/ppl. {@link ProfileScope} calls {@link #onPhaseStart(String)} when a
+ * phase opens and closes the returned handle when the phase ends, so the listener sees the exact
+ * same start/end boundary that feeds the profile metric.
+ *
+ * <p>The default implementation is a no-op ({@link #NOOP}). Modules that emit tracing spans (e.g.
+ * the {@code opensearch} module) install a concrete listener via {@link
+ * ProfileScope#installListener(PhaseListener)}.
+ */
+public interface PhaseListener {
+
+  /** Called when a phase starts. The returned handle is closed when the phase ends. */
+  Handle onPhaseStart(String phaseName);
+
+  /** Observation of a single phase — the caller closes it (and may call {@link #setError}). */
+  interface Handle extends AutoCloseable {
+    /** Record a failure that occurred inside the phase. Called before {@link #close()}. */
+    default void setError(Throwable t) {}
+
+    @Override
+    void close();
+  }
+
+  Handle NOOP_HANDLE =
+      new Handle() {
+        @Override
+        public void close() {}
+      };
+
+  PhaseListener NOOP = phaseName -> NOOP_HANDLE;
+}

core/src/main/java/org/opensearch/sql/monitor/profile/ProfileScope.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.monitor.profile;
+
+import java.util.Locale;
+
+/**
+ * One measured boundary that feeds both query profiling and — if a {@link PhaseListener} is
+ * installed — distributed tracing. Times a code block with {@link System#nanoTime()} and, on {@link
+ * #close()}, adds the elapsed nanoseconds to the phase's {@link ProfileMetric} and closes the
+ * listener's handle. Use with try-with-resources so both outputs share the same start/end boundary.
+ *
+ * <p>The phase name (span name and profile phase key alike) is derived from the {@link MetricName}
+ * — {@code metric.name().toLowerCase(Locale.ROOT)} — the same rule {@link QueryProfile} uses, so
+ * the two cannot drift apart.
+ *
+ * <p>Callers call {@link #setError(Throwable)} before rethrowing to record a failure — the listener
+ * forwards it to its handle (e.g. onto a span). try-with-resources cannot observe the thrown
+ * exception here.
+ *
+ * <p>Kept trace-free by design: the {@link PhaseListener} indirection means core/ppl never
+ * reference {@code Tracer}. The {@code opensearch} module installs the tracing listener at startup.
+ */
+public final class ProfileScope implements AutoCloseable {
+
+  private static volatile PhaseListener listener = PhaseListener.NOOP;
+
+  /** Install a global phase listener (typically from the {@code opensearch} module). */
+  public static void installListener(PhaseListener newListener) {
+    listener = newListener == null ? PhaseListener.NOOP : newListener;
+  }
+
+  private final ProfileMetric metric;
+  private final long startNanos;
+  private final PhaseListener.Handle handle;
+
+  private ProfileScope(ProfileMetric metric, long startNanos, PhaseListener.Handle handle) {
+    this.metric = metric;
+    this.startNanos = startNanos;
+    this.handle = handle;
+  }
+
+  /** Open a phase that records a profile metric and (if a listener is installed) a tracing span. */
+  public static ProfileScope open(MetricName metric) {
+    return new ProfileScope(
+        QueryProfiling.current().getOrCreateMetric(metric),
+        System.nanoTime(),
+        listener.onPhaseStart(metric.name().toLowerCase(Locale.ROOT)));
+  }
+
+  /**
+   * Open a trace-only phase — a span with no matching profile metric. Used for work that happens
+   * before {@link QueryProfiling} is activated on the current thread (e.g. transport-side parse and
+   * dispatch) and therefore can't feed a metric anyway. {@code phaseName} must be a fixed literal
+   * chosen by the caller, not derived from user input.
+   */
+  public static ProfileScope openTraceOnly(String phaseName) {
+    return new ProfileScope(
+        NoopProfileMetric.INSTANCE, System.nanoTime(), listener.onPhaseStart(phaseName));
+  }
+
+  /** Record a failure on the listener's handle. Callers invoke this before rethrowing. */
+  public void setError(Throwable t) {
+    handle.setError(t);
+  }
+
+  @Override
+  public void close() {
+    metric.add(System.nanoTime() - startNanos);
+    handle.close();
+  }
+}