/*
   Copyright (c) 2011 James Ahlborn
   
   Licensed 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 com.healthmarketscience.jackcess.impl;
  
  import java.io.IOException;
  import java.io.UncheckedIOException;
  import java.lang.System.Logger;
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.Map;
  import java.util.Set;
  
  import com.healthmarketscience.jackcess.Index;
  import com.healthmarketscience.jackcess.IndexCursor;
  import com.healthmarketscience.jackcess.Row;
  import com.healthmarketscience.jackcess.impl.TableImpl.RowState;
  import com.healthmarketscience.jackcess.util.CaseInsensitiveColumnMatcher;
  import com.healthmarketscience.jackcess.util.ColumnMatcher;
  import com.healthmarketscience.jackcess.util.EntryIterableBuilder;
  import com.healthmarketscience.jackcess.util.SimpleColumnMatcher;
  
  /**
   * Cursor backed by an index with extended traversal options.
   *
   * @author James Ahlborn
   */
  public class IndexCursorImpl extends CursorImpl implements IndexCursor
  {
    private static final Logger LOG = System.getLogger(IndexCursorImpl.class.getName());
  
    /** IndexDirHandler for forward traversal */
    private final IndexDirHandler _forwardDirHandler =
      new ForwardIndexDirHandler();
    /** IndexDirHandler for backward traversal */
    private final IndexDirHandler _reverseDirHandler =
      new ReverseIndexDirHandler();
    /** logical index which this cursor is using */
    private final IndexImpl _index;
    /** Cursor over the entries of the relevant index */
    private final IndexData.EntryCursor _entryCursor;
    /** column names for the index entry columns */
    private Set<String> _indexEntryPattern;
  
    private IndexCursorImpl(TableImpl table, IndexImpl index,
                            IndexData.EntryCursor entryCursor)
      throws IOException
    {
      super(new IdImpl(table, index), table,
            new IndexPosition(entryCursor.getFirstEntry()),
            new IndexPosition(entryCursor.getLastEntry()));
      _index = index;
      _index.initialize();
      _entryCursor = entryCursor;
    }
  
    /**
     * Creates an indexed cursor for the given table, narrowed to the given
     * range.
     * <p>
     * Note, index based table traversal may not include all rows, as certain
     * types of indexes do not include all entries (namely, some indexes ignore
     * null entries, see {@link Index#shouldIgnoreNulls}).
     *
     * @param table the table over which this cursor will traverse
     * @param index index for the table which will define traversal order as
     *              well as enhance certain lookups
     * @param startRow the first row of data for the cursor, or {@code null} for
     *                 the first entry
     * @param startInclusive whether or not startRow is inclusive or exclusive
     * @param endRow the last row of data for the cursor, or {@code null} for
     *               the last entry
     * @param endInclusive whether or not endRow is inclusive or exclusive
     */
    public static IndexCursorImpl createCursor(TableImpl table, IndexImpl index,
                                               Object[] startRow,
                                               boolean startInclusive,
                                               Object[] endRow,
                                               boolean endInclusive)
      throws IOException
    {
      if(table != index.getTable()) {
        throw new IllegalArgumentException(
            "Given index is not for given table: " + index + ", " + table);
     }
     if(!index.getIndexData().isValid()) {
       throw new IllegalArgumentException(
           "Given index " + index +
           " is not usable for indexed lookups due to " +
           index.getIndexData().getUnsupportedReason());
     }
     IndexCursorImpl/IndexCursorImpl.html#IndexCursorImpl">IndexCursorImpl cursor = new IndexCursorImpl(
         table, index, index.cursor(startRow, startInclusive,
                                    endRow, endInclusive));
     // init the column matcher appropriately for the index type
     cursor.setColumnMatcher(null);
     return cursor;
   }
 
   private Set<String> getIndexEntryPattern()
   {
     if(_indexEntryPattern == null) {
       // init our set of index column names
       _indexEntryPattern = new HashSet<>();
       for(IndexData.ColumnDescriptor col : getIndex().getColumns()) {
         _indexEntryPattern.add(col.getName());
       }
     }
     return _indexEntryPattern;
   }
 
   @Override
   public IndexImpl getIndex() {
     return _index;
   }
 
   @Override
   public Row findRowByEntry(Object... entryValues)
     throws IOException
   {
     if(findFirstRowByEntry(entryValues)) {
       return getCurrentRow();
     }
     return null;
   }
 
   @Override
   public boolean findFirstRowByEntry(Object... entryValues)
     throws IOException
   {
     PositionImpl curPos = _curPos;
     PositionImpl prevPos = _prevPos;
     boolean found = false;
     try {
       found = findFirstRowByEntryImpl(toRowValues(entryValues), true,
                                       _columnMatcher);
       return found;
     } finally {
       if(!found) {
         try {
           restorePosition(curPos, prevPos);
         } catch(IOException e) {
           LOG.log(Logger.Level.ERROR, "Failed restoring position", e);
         }
       }
     }
   }
 
   @Override
   public void findClosestRowByEntry(Object... entryValues)
     throws IOException
   {
     PositionImpl curPos = _curPos;
     PositionImpl prevPos = _prevPos;
     boolean found = false;
     try {
       findFirstRowByEntryImpl(toRowValues(entryValues), false,
                               _columnMatcher);
       found = true;
     } finally {
       if(!found) {
         try {
           restorePosition(curPos, prevPos);
         } catch(IOException e) {
           LOG.log(Logger.Level.ERROR, "Failed restoring position", e);
         }
       }
     }
   }
 
   @Override
   public boolean currentRowMatchesEntry(Object... entryValues)
     throws IOException
   {
     return currentRowMatchesEntryImpl(toRowValues(entryValues), _columnMatcher);
   }
 
   @Override
   public EntryIterableBuilder newEntryIterable(Object... entryValues) {
     return new EntryIterableBuilder(this, entryValues);
   }
 
   public Iterator<Row> entryIterator(EntryIterableBuilder iterBuilder) {
     return new EntryIterator(iterBuilder.getColumnNames(),
                              toRowValues(iterBuilder.getEntryValues()),
                              iterBuilder.getColumnMatcher());
   }
 
   @Override
   protected IndexDirHandler getDirHandler(boolean moveForward) {
     return (moveForward ? _forwardDirHandler : _reverseDirHandler);
   }
 
   @Override
   protected boolean isUpToDate() {
     return(super.isUpToDate() && _entryCursor.isUpToDate());
   }
 
   @Override
   protected void reset(boolean moveForward) {
     _entryCursor.reset(moveForward);
     super.reset(moveForward);
   }
 
   @Override
   protected void restorePositionImpl(PositionImpl curPos, PositionImpl prevPos)
     throws IOException
   {
     if(!(curPos instanceof IndexPosition) ||
        !(prevPos instanceof IndexPosition)) {
       throw new IllegalArgumentException(
           "Restored positions must be index positions");
     }
     _entryCursor.restorePosition(((IndexPosition)curPos).getEntry(),
                                  ((IndexPosition)prevPos).getEntry());
     super.restorePositionImpl(curPos, prevPos);
   }
 
   @Override
   protected PositionImpl getRowPosition(RowIdImpl rowId) throws IOException
   {
     // we need to get the index entry which corresponds with this row
     Row row = getTable().getRow(getRowState(), rowId, getIndexEntryPattern());
     _entryCursor.beforeEntry(getTable().asRow(row));
     return new IndexPosition(_entryCursor.getNextEntry());
   }
 
   @Override
   protected boolean findAnotherRowImpl(
       ColumnImpl columnPattern, Object valuePattern, boolean moveForward,
       ColumnMatcher columnMatcher, Object searchInfo)
     throws IOException
   {
     Object[] rowValues = (Object[])searchInfo;
 
     if((rowValues == null) || !isAtBeginning(moveForward)) {
       // use the default table scan if we don't have index data or we are
       // mid-cursor
       return super.findAnotherRowImpl(columnPattern, valuePattern, moveForward,
                                       columnMatcher, rowValues);
     }
 
     // sweet, we can use our index
     if(!findPotentialRow(rowValues, true)) {
       return false;
     }
 
     return findMatchingRowInRun(
         rowValues, moveForward, columnMatcher,
         () -> currentRowMatchesImpl(columnPattern, valuePattern,
                                     columnMatcher));
   }
 
   /**
    * Moves to the first row (as defined by the cursor) where the index entries
    * match the given values.  Caller manages save/restore on failure.
    *
    * @param rowValues the column values built from the index column values
    * @param requireMatch whether or not an exact match is desired
    * @return {@code true} if a valid row was found with the given values,
    *         {@code false} if no row was found
    */
   protected boolean findFirstRowByEntryImpl(Object[] rowValues,
                                             boolean requireMatch,
                                             ColumnMatcher columnMatcher)
     throws IOException
   {
     if(!findPotentialRow(rowValues, requireMatch)) {
       return false;
     } else if(!requireMatch) {
       // nothing more to do, we have moved to the closest row
       return true;
     }
 
     return findMatchingRowInRun(rowValues, MOVE_FORWARD, columnMatcher,
                                 MATCHES_ENTRY);
   }
 
   @Override
   protected boolean findAnotherRowImpl(
       Map<String,?> rowPattern, boolean moveForward,
       ColumnMatcher columnMatcher, Object searchInfo)
     throws IOException
   {
     Object[] rowValues = (Object[])searchInfo;
 
     if((rowValues == null) || !isAtBeginning(moveForward)) {
       // use the default table scan if we don't have index data or we are
       // mid-cursor
       return super.findAnotherRowImpl(rowPattern, moveForward, columnMatcher,
                                       rowValues);
     }
 
     // sweet, we can use our index
     if(!findPotentialRow(rowValues, true)) {
       // at end of index, no potential matches
       return false;
     }
 
     // determine if the pattern columns exactly match the index columns
     boolean exactColumnMatch = rowPattern.keySet().equals(
         getIndexEntryPattern());
 
     // note, if exactColumnMatch, no need to do an extra comparison with the
     // current row (since the entry match check in the run is equivalent to
     // this check)
     return findMatchingRowInRun(
         rowValues, moveForward, columnMatcher,
         () -> (exactColumnMatch ||
                currentRowMatchesImpl(rowPattern, columnMatcher)));
   }
 
   /**
    * A test of the row the cursor is on, run over the rows which could match a
    * search.
    */
   private interface RowTest
   {
     public boolean matches() throws IOException;
   }
 
   /** the test for a search which asks nothing beyond the index entry */
   private static final RowTest MATCHES_ENTRY = () -> true;
 
   /**
    * Returns whether any of the rows which could still match the search passes
    * the given test, leaving the cursor on the first which does.
    * <p>
    * The cursor starts on the row the search landed on, which is the first of
    * the rows sharing its index entry.
    */
   private boolean findMatchingRowInRun(Object[] rowValues, boolean moveForward,
                                        ColumnMatcher columnMatcher,
                                        RowTest test)
     throws IOException
   {
     byte[] runBytes = currentEntryBytes();
 
     do {
 
       if(currentRowMatchesEntryImpl(rowValues, columnMatcher) &&
          test.matches()) {
         return true;
       }
 
     } while(moveToAnotherRow(moveForward) &&
             couldStillMatch(rowValues, columnMatcher, runBytes));
 
     return false;
   }
 
   /**
    * Returns whether the row the cursor is on can still match the search.
    */
   private boolean couldStillMatch(Object[] rowValues,
                                   ColumnMatcher columnMatcher,
                                   byte[] runBytes)
     throws IOException
   {
     if(currentRowMatchesEntryImpl(rowValues, columnMatcher)) {
       return true;
     }
 
     // the values differ, but one index entry can hold several rows, since the
     // entry of a text value folds case, trailing spaces and the characters
     // the collation treats alike.  the rest of that run is still worth
     // reading, and a row past it has a greater entry and so cannot hold the
     // values searched for
     byte[] entryBytes = currentEntryBytes();
     return ((runBytes != null) && Arrays.equals(runBytes, entryBytes));
   }
 
   /**
    * Returns the entry bytes of the row the cursor is on, or {@code null} if
    * it is not on a row.
    */
   private byte[] currentEntryBytes() {
     IndexData.Entry entry = ((IndexPosition)_curPos).getEntry();
     return (entry.isValid() ? entry.getEntryBytes() : null);
   }
 
   private boolean currentRowMatchesEntryImpl(Object[] rowValues,
                                              ColumnMatcher columnMatcher)
     throws IOException
   {
     // check the next row to see if it actually matches
     Row row = getCurrentRow(getIndexEntryPattern());
 
     for(IndexData.ColumnDescriptor col : getIndex().getColumns()) {
 
       Object patValue = rowValues[col.getColumnIndex()];
 
       if((patValue == IndexData.MIN_VALUE) ||
          (patValue == IndexData.MAX_VALUE)) {
         // all remaining entry values are "special" (used for partial lookups)
         return true;
       }
 
       String columnName = col.getName();
       Object rowValue = row.get(columnName);
       if(!columnMatcher.matches(getTable(), columnName, patValue, rowValue)) {
         return false;
       }
     }
 
     return true;
   }
 
   private boolean findPotentialRow(Object[] rowValues, boolean requireMatch)
     throws IOException
   {
     _entryCursor.beforeEntry(rowValues);
     IndexData.Entry startEntry = _entryCursor.getNextEntry();
     if(requireMatch && !startEntry.getRowId().isValid()) {
       // at end of index, no potential matches
       return false;
     }
     // move to position and check it out
     restorePosition(new IndexPosition(startEntry));
     return true;
   }
 
   @Override
   protected Object prepareSearchInfo(ColumnImpl columnPattern, Object valuePattern)
   {
     // attempt to generate a lookup row for this index
     return _entryCursor.getIndexData().constructPartialIndexRow(
         IndexData.MIN_VALUE, columnPattern.getName(), valuePattern);
   }
 
   @Override
   protected Object prepareSearchInfo(Map<String,?> rowPattern)
   {
     // attempt to generate a lookup row for this index
     return _entryCursor.getIndexData().constructPartialIndexRow(
         IndexData.MIN_VALUE, rowPattern);
   }
 
   @Override
   protected boolean keepSearching(ColumnMatcher columnMatcher,
                                   Object searchInfo)
     throws IOException
   {
     if(searchInfo instanceof Object[]) {
       // if we have a lookup row for this index, then we only need to continue
       // searching while we are looking at rows which match the index lookup
       // value(s).  once we move past those rows, no other rows could possibly
       // match.
       return currentRowMatchesEntryImpl((Object[])searchInfo, columnMatcher);
     }
     // we are doing a full table scan
     return true;
   }
 
   private Object[] toRowValues(Object[] entryValues)
   {
     return _entryCursor.getIndexData().constructPartialIndexRowFromEntry(
         IndexData.MIN_VALUE, entryValues);
   }
 
   @Override
   protected PositionImpl findAnotherPosition(
       RowState rowState, PositionImpl curPos, boolean moveForward)
     throws IOException
   {
     IndexDirHandler handler = getDirHandler(moveForward);
     IndexPosition endPos = (IndexPosition)handler.getEndPosition();
     IndexData.Entry entry = handler.getAnotherEntry();
     return ((!entry.equals(endPos.getEntry())) ?
             new IndexPosition(entry) : endPos);
   }
 
   @Override
   protected ColumnMatcher getDefaultColumnMatcher() {
     if(getIndex().isUnique()) {
       // text indexes are case-insensitive, therefore we should always use a
       // case-insensitive matcher for unique indexes.
       return CaseInsensitiveColumnMatcher.INSTANCE;
     }
     return SimpleColumnMatcher.INSTANCE;
   }
 
   /**
    * Handles moving the table index cursor in a given direction.  Separates
    * cursor logic from value storage.
    */
   private abstract class IndexDirHandler extends DirHandler {
     public abstract IndexData.Entry getAnotherEntry()
       throws IOException;
   }
 
   /**
    * Handles moving the table index cursor forward.
    */
   private final class ForwardIndexDirHandler extends IndexDirHandler {
     @Override
     public PositionImpl getBeginningPosition() {
       return getFirstPosition();
     }
     @Override
     public PositionImpl getEndPosition() {
       return getLastPosition();
     }
     @Override
     public IndexData.Entry getAnotherEntry() throws IOException {
       return _entryCursor.getNextEntry();
     }
   }
 
   /**
    * Handles moving the table index cursor backward.
    */
   private final class ReverseIndexDirHandler extends IndexDirHandler {
     @Override
     public PositionImpl getBeginningPosition() {
       return getLastPosition();
     }
     @Override
     public PositionImpl getEndPosition() {
       return getFirstPosition();
     }
     @Override
     public IndexData.Entry getAnotherEntry() throws IOException {
       return _entryCursor.getPreviousEntry();
     }
   }
 
   /**
    * Value object which maintains the current position of an IndexCursor.
    */
   private static final class IndexPosition extends PositionImpl
   {
     private final IndexData.Entry _entry;
 
     private IndexPosition(IndexData.Entry entry) {
       _entry = entry;
     }
 
     @Override
     public RowIdImpl getRowId() {
       return getEntry().getRowId();
     }
 
     public IndexData.Entry getEntry() {
       return _entry;
     }
 
     @Override
     protected boolean equalsImpl(Object o) {
       return getEntry().equals(((IndexPosition)o).getEntry());
     }
 
     @Override
     public String toString() {
       return "Entry = " + getEntry();
     }
   }
 
   /**
    * Row iterator (by matching entry) for this cursor, modifiable.
    */
   private final class EntryIterator extends BaseIterator
   {
     private final Object[] _rowValues;
 
     private EntryIterator(Collection<String> columnNames, Object[] rowValues,
                           ColumnMatcher columnMatcher)
     {
       super(columnNames, false, MOVE_FORWARD, columnMatcher);
       _rowValues = rowValues;
       try {
         _hasNext = findFirstRowByEntryImpl(rowValues, true, _columnMatcher);
         _validRow = _hasNext;
       } catch(IOException e) {
           throw new UncheckedIOException(e);
       }
     }
 
     @Override
     protected boolean findNext() throws IOException {
       // the next row may share its entry with one which does not match, so
       // the rest of that run is read before the iteration ends
       return (moveToNextRow() &&
               findMatchingRowInRun(_rowValues, MOVE_FORWARD, _colMatcher,
                                    MATCHES_ENTRY));
     }
   }
 
 }