docs: public void emitLogsAndMetrics(
        @Nullable final Throwable e,
        @Nullable final String remoteAddress,
        final long bytesWritten
    )
    {
      synchronized (lock) {
        if (sql == null) {
          // Never initialized, don't log or emit anything.
          return;
        }
  
        if (state == State.DONE) {
          log.warn("Tried to emit logs and metrics twice for query[%s]!", sqlQueryId());
        }
  
        state = State.DONE;
  
        final boolean success = e == null;
        final long queryTimeNs = System.nanoTime() - startNs;
  
        try {
          ServiceMetricEvent.Builder metricBuilder = ServiceMetricEvent.builder();
          if (plannerContext != null) {
            metricBuilder.setDimension("id", plannerContext.getSqlQueryId());
            metricBuilder.setDimension("nativeQueryIds", plannerContext.getNativeQueryIds().toString());
          }
          if (plannerResult != null) {
            metricBuilder.setDimension("dataSource", plannerResult.datasourceNames().toString());
          }
          metricBuilder.setDimension("remoteAddress", StringUtils.nullToEmptyNonDruidDataString(remoteAddress));
          metricBuilder.setDimension("success", String.valueOf(success));
          emitter.emit(metricBuilder.build("sqlQuery/time", TimeUnit.NANOSECONDS.toMillis(queryTimeNs)));
          if (bytesWritten >= 0) {
            emitter.emit(metricBuilder.build("sqlQuery/bytes", bytesWritten));
          }
  
          final Map<String, Object> statsMap = new LinkedHashMap<>();
          statsMap.put("sqlQuery/time", TimeUnit.NANOSECONDS.toMillis(queryTimeNs));
          statsMap.put("sqlQuery/bytes", bytesWritten);
          statsMap.put("success", success);
          statsMap.put("context", queryContext);
          if (plannerContext != null) {
            statsMap.put("identity", plannerContext.getAuthenticationResult().getIdentity());
            queryContext.put("nativeQueryIds", plannerContext.getNativeQueryIds().toString());
          }
          if (e != null) {
            statsMap.put("exception", e.toString());
  
            if (e instanceof QueryInterruptedException) {
              statsMap.put("interrupted", true);
              statsMap.put("reason", e.toString());
            }
          }
  
          requestLogger.logSqlQuery(
              RequestLogLine.forSql(
                  sql,
                  queryContext,
                  DateTimes.utc(startMs),
                  remoteAddress,
                  new QueryStats(statsMap)
              )
          );
        }
        catch (Exception ex) {
          log.error(ex, "Unable to log sql [%s]!", sql);
        }
      }
    }