content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to you under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   * http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.calcite.runtime;
  
  import org.apache.calcite.interpreter.Row;
  import org.apache.calcite.linq4j.AbstractEnumerable;
  import org.apache.calcite.linq4j.Enumerable;
  import org.apache.calcite.linq4j.Enumerator;
  import org.apache.calcite.linq4j.function.Function1;
  
  import org.checkerframework.checker.nullness.qual.Nullable;
  
  import java.util.ArrayDeque;
  import java.util.Deque;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.function.Consumer;
  import java.util.function.Supplier;
  
  import static java.util.Objects.requireNonNull;
  
  /**
   * Utilities for processing {@link org.apache.calcite.linq4j.Enumerable}
   * collections.
   *
   * <p>This class is a place to put things not yet added to linq4j.
   * Methods are subject to removal without notice.
   */
  public class Enumerables {
  
    private Enumerables() {}
  
    /** Converts an enumerable over singleton arrays into the enumerable of their
     * first elements. */
    public static <E> Enumerable<E> slice0(Enumerable<E[]> enumerable) {
      return enumerable.select(elements -> elements[0]);
    }
  
    /** Converts an {@link Enumerable} over object arrays into an
     * {@link Enumerable} over {@link Row} objects. */
    public static Enumerable<Row> toRow(final Enumerable<@Nullable Object[]> enumerable) {
      return enumerable.select((Function1<@Nullable Object[], Row>) Row::asCopy);
    }
  
    /** Converts a supplier of an {@link Enumerable} over object arrays into a
     * supplier of an {@link Enumerable} over {@link Row} objects. */
    public static Supplier<Enumerable<Row>> toRow(
        final Supplier<Enumerable<@Nullable Object[]>> supplier) {
      return () -> toRow(supplier.get());
    }
  
    @SuppressWarnings("Guava")
    @Deprecated // to be removed before 2.0
    public static com.google.common.base.Supplier<Enumerable<Row>> toRow(
        final com.google.common.base.Supplier<Enumerable<@Nullable Object[]>> supplier) {
      return () -> toRow(supplier.get());
    }
  
    public static <E, TKey, TResult> Enumerable<TResult> match(
        Enumerable<E> enumerable,
        final Function1<E, TKey> keySelector,
        Matcher<E> matcher,
        Emitter<E, TResult> emitter, int history, int future) {
      return new AbstractEnumerable<TResult>() {
        @Override public Enumerator<TResult> enumerator() {
          return new Enumerator<TResult>() {
            final Enumerator<E> inputEnumerator = enumerable.enumerator();
  
            // State of each partition.
            final Map<TKey, Matcher.PartitionState<E>> partitionStates =
                new HashMap<>();
  
            int inputRow = -1;
  
            final Deque<TResult> emitRows = new ArrayDeque<>();
  
            /** Current result row. Null if no row is ready. */
            @Nullable TResult resultRow;
  
            /** Match counter is 1-based in Oracle. */
            final AtomicInteger matchCounter = new AtomicInteger(1);
  
            @Override public TResult current() {
              requireNonNull(resultRow, "resultRow");
              return resultRow;
            }
  
            @Override public boolean moveNext() {
              for (;;) {
                resultRow = emitRows.pollFirst();
                if (resultRow != null) {
                  return true;
                }
                // No rows are currently read to emit. Read the next input row,
                // see whether it completes a match (or matches), and if so, add
                // the resulting rows to the buffer.
                if (!inputEnumerator.moveNext()) {
                  return false;
                }
                ++inputRow;
                final E row = inputEnumerator.current();
                final TKey key = keySelector.apply(row);
                final Matcher.PartitionState<E> partitionState =
                    partitionStates.computeIfAbsent(key,
                        k -> matcher.createPartitionState(history, future));
  
                partitionState.getMemoryFactory().add(row);
  
  
                matcher.matchOne(partitionState.getRows(), partitionState,
                    // TODO 26.12.18 jf: add row states (whatever this is?)
                    matches -> emitter.emit(matches.rows, null, matches.symbols,
                        matchCounter.getAndIncrement(), emitRows::add));
  /*
                recentRows.add(e);
                int earliestRetainedRow = Integer.MAX_VALUE;
                for (int i = 0; i < partitionState.incompleteMatches.size(); i++) {
                  MatchState match = partitionState.incompleteMatches.get(i);
                  earliestRetainedRow = Math.min(earliestRetainedRow, match.firstRow);
                  final int state = automaton.nextState(match.state, e);
                  switch (state) {
                  case Automaton.ACCEPT:
                    final List<E> matchedRows =
                        recentRows.subList(0, 0); // TODO:
                    final List<Integer> rowStates = ImmutableList.of(); // TODO:
                    emitRows.addAll(
                        emitter.emit(matchedRows, rowStates,
                            partitionState.matchCount++));
                    // fall through
                  case Automaton.FAIL:
                    partitionState.incompleteMatches.remove(i--);
                    break;
                  default:
                    match.state = state;
                  }
                }
                // Try to start a match based on the current row
                final int state = automaton.nextState(Automaton.START_STATE, e);
                switch (state) {
                case Automaton.ACCEPT:
                  final List<E> matchedRows = ImmutableList.of(e);
                  final List<Integer> rowStates = ImmutableList.of(state);
                  emitRows.addAll(
                      emitter.emit(matchedRows, rowStates,
                          partitionState.matchCount++));
                  // fall through
                case Automaton.FAIL:
                  // since it immediately succeeded or failed, don't add
                  // it to the queue
                  break;
                default:
                  partitionState.incompleteMatches.add(
                      new MatchState(inputRow, state));
                }
  */
              }
            }
  
            @Override public void reset() {
              throw new UnsupportedOperationException();
            }
  
            @Override public void close() {
              inputEnumerator.close();
            }
          };
        }
      };
    }
  
    /** Given a match (a list of rows, and their states) produces a list
     * of rows to be output.
     *
     * @param <E> element type
     * @param <TResult> result type */
    public interface Emitter<E, TResult> {
      void emit(List<E> rows, @Nullable List<Integer> rowStates, List<String> rowSymbols, int match,
          Consumer<TResult> consumer);
    }
  }
  