@SuppressWarnings({"BooleanExpressionComplexity", "CyclomaticComplexity"})
  public static boolean isScalablePushQuery(
      final Statement statement,
      final KsqlExecutionContext ksqlEngine,
      final KsqlConfig ksqlConfig,
      final Map<String, Object> overrides
  ) {
    if (!isPushV2Enabled(ksqlConfig, overrides)) {
      return false;
    }
    if (! (statement instanceof Query)) {
      return false;
    }
    final Query query = (Query) statement;
    final SourceFinder sourceFinder = new SourceFinder();
    sourceFinder.process(query.getFrom(), null);
    // It will be present if it's not a join, which we don't handle
    if (!sourceFinder.getSourceName().isPresent()) {
      return false;
    }
    // Find all of the writers to this particular source.
    final SourceName sourceName = sourceFinder.getSourceName().get();
    final Set<QueryId> upstreamQueries = ksqlEngine.getQueriesWithSink(sourceName);
    // See if the config or override have set the stream to be "latest"
    final boolean isLatest = isLatest(ksqlConfig, overrides);
    // Cannot be a pull query, i.e. must be a push
    return !query.isPullQuery()
        // Group by is not supported
        && !query.getGroupBy().isPresent()
        // Windowing is not supported
        && !query.getWindow().isPresent()
        // Having clause is not supported
        && !query.getHaving().isPresent()
        // Partition by is not supported
        && !query.getPartitionBy().isPresent()
        // There must be an EMIT CHANGES clause
        && (query.getRefinement().isPresent()
            && query.getRefinement().get().getOutputRefinement() == OutputRefinement.CHANGES)
        // Must be reading from "latest"
        && isLatest
        // We only handle a single sink source at the moment from a CTAS/CSAS
        && upstreamQueries.size() == 1
        // ROWPARTITION and ROWOFFSET are not currently supported in SPQs
        && !containsDisallowedColumns(query);
  }
@Test
  public void isScalablePushQuery_false_hasNoRefinement() {
    try(MockedStatic<ColumnExtractor> columnExtractor = mockStatic(ColumnExtractor.class)) {
      // When:
      expectIsSPQ(ColumnName.of("foo"), columnExtractor);
      when(query.getRefinement()).thenReturn(Optional.empty());

      // Then:
      assertThat(ScalablePushUtil.isScalablePushQuery(query, ksqlEngine, ksqlConfig,
          ImmutableMap.of(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")),
          equalTo(false));
    }
  }