@SuppressWarnings("unchecked")
  public QueryMetadataHolder handleStatement(
      final ServiceContext serviceContext,
      final Map<String, Object> configOverrides,
      final Map<String, Object> requestProperties,
      final PreparedStatement<?> statement,
      final Optional<Boolean> isInternalRequest,
      final MetricsCallbackHolder metricsCallbackHolder,
      final Context context,
      final boolean excludeTombstones
  ) {

    if (statement.getStatement() instanceof Query) {
      return handleQuery(
          serviceContext,
          (PreparedStatement<Query>) statement,
          isInternalRequest,
          metricsCallbackHolder,
          configOverrides,
          requestProperties,
          context,
          excludeTombstones
      );
    } else {
      return QueryMetadataHolder.unhandled();
    }
  }
@Test
  public void queryLoggerShouldNotReceiveStatementsWhenHandlePullQuery() {
    when(mockDataSource.getDataSourceType()).thenReturn(DataSourceType.KTABLE);
    when(ksqlEngine.executeTablePullQuery(any(), any(), any(), any(), any(), any(), any(),
        anyBoolean(), any()))
        .thenReturn(pullQueryResult);
    try (MockedStatic<QueryLogger> logger = Mockito.mockStatic(QueryLogger.class)) {
      queryExecutor.handleStatement(serviceContext, ImmutableMap.of(), ImmutableMap.of(),
          pullQuery, Optional.empty(), metricsCallbackHolder, context, false);

      logger.verify(() -> QueryLogger.info("Transient query created",
          PULL_QUERY_STRING), never());
    }
  }