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.cassandra.cql3.selection;
  
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.util.Arrays;
  import java.util.List;
  
  import org.apache.cassandra.cql3.ColumnIdentifier;
  import org.apache.cassandra.cql3.ColumnSpecification;
  import org.apache.cassandra.cql3.QueryOptions;
  import org.apache.cassandra.cql3.functions.Function;
  import org.apache.cassandra.cql3.selection.ColumnTimestamps.TimestampsType;
  import org.apache.cassandra.db.TypeSizes;
  import org.apache.cassandra.db.context.CounterContext;
  import org.apache.cassandra.db.filter.ColumnFilter;
  import org.apache.cassandra.db.marshal.AbstractType;
  import org.apache.cassandra.db.marshal.CollectionType;
  import org.apache.cassandra.db.marshal.UserType;
  import org.apache.cassandra.db.rows.Cell;
  import org.apache.cassandra.db.rows.ColumnData;
  import org.apache.cassandra.db.rows.ComplexColumnData;
  import org.apache.cassandra.exceptions.InvalidRequestException;
  import org.apache.cassandra.io.util.DataInputPlus;
  import org.apache.cassandra.io.util.DataOutputPlus;
  import org.apache.cassandra.schema.CQLTypeParser;
  import org.apache.cassandra.schema.ColumnMetadata;
  import org.apache.cassandra.schema.KeyspaceMetadata;
  import org.apache.cassandra.schema.Schema;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.transport.ProtocolVersion;
  import org.apache.cassandra.utils.ByteBufferUtil;
  
  import static org.apache.cassandra.cql3.statements.RequestValidations.invalidRequest;
  
  /**
   * A <code>Selector</code> is used to convert the data returned by the storage engine into the data requested by the
   * user. They correspond to the &lt;selector&gt; elements from the select clause.
   * <p>Since the introduction of aggregation, <code>Selector</code>s cannot be called anymore by multiple threads
   * as they have an internal state.</p>
   */
  public abstract class Selector
  {
      protected static abstract class SelectorDeserializer
      {
          protected abstract Selector deserialize(DataInputPlus in, int version, TableMetadata metadata) throws IOException;
  
          protected final AbstractType<?> readType(TableMetadata metadata, DataInputPlus in) throws IOException
          {
              KeyspaceMetadata keyspace = Schema.instance.getKeyspaceMetadata(metadata.keyspace);
              return readType(keyspace, in);
          }
  
          protected final AbstractType<?> readType(KeyspaceMetadata keyspace, DataInputPlus in) throws IOException
          {
              String cqlType = in.readUTF();
              return CQLTypeParser.parse(keyspace.name, cqlType, keyspace.types);
          }
      }
  
      /**
       * The <code>Selector</code> kinds.
       */
      public enum Kind
      {
          SIMPLE_SELECTOR(SimpleSelector.deserializer),
          TERM_SELECTOR(TermSelector.deserializer),
          WRITETIME_OR_TTL_SELECTOR(WritetimeOrTTLSelector.deserializer),
          LIST_SELECTOR(ListSelector.deserializer),
          SET_SELECTOR(SetSelector.deserializer),
          MAP_SELECTOR(MapSelector.deserializer),
          TUPLE_SELECTOR(TupleSelector.deserializer),
          USER_TYPE_SELECTOR(UserTypeSelector.deserializer),
          FIELD_SELECTOR(FieldSelector.deserializer),
          SCALAR_FUNCTION_SELECTOR(ScalarFunctionSelector.deserializer),
          AGGREGATE_FUNCTION_SELECTOR(AggregateFunctionSelector.deserializer),
          ELEMENT_SELECTOR(ElementsSelector.ElementSelector.deserializer),
          SLICE_SELECTOR(ElementsSelector.SliceSelector.deserializer),
          VECTOR_SELECTOR(VectorSelector.deserializer);
  
          private final SelectorDeserializer deserializer;
  
          Kind(SelectorDeserializer deserializer)
          {
              this.deserializer = deserializer;
          }
      }
  
      /**
       * A factory for <code>Selector</code> instances.
       */
      public static abstract class Factory
      {
          public void addFunctionsTo(List<Function> functions)
          {
          }
  
          /**
           * Returns the column specification corresponding to the output value of the selector instances created by
           * this factory.
           *
           * @param table the table meta data
           * @return a column specification
           */
          public ColumnSpecification getColumnSpecification(TableMetadata table)
          {
              return new ColumnSpecification(table.keyspace,
                                             table.name,
                                             new ColumnIdentifier(getColumnName(), true), // note that the name is not necessarily
                                                                                          // a true column name so we shouldn't intern it
                                             getReturnType());
          }
  
          /**
           * Creates a new <code>Selector</code> instance.
           *
           * @param options the options of the query for which the instance is created (some selector
           * depends on the bound values in particular).
           * @return a new <code>Selector</code> instance
           */
          public abstract Selector newInstance(QueryOptions options);
  
          /**
           * Checks if this factory creates selectors instances that creates aggregates.
           *
           * @return <code>true</code> if this factory creates selectors instances that creates aggregates,
           * <code>false</code> otherwise
           */
          public boolean isAggregateSelectorFactory()
          {
              return false;
          }
  
          /**
           * Checks if this factory creates <code>writetime</code> selectors instances.
           *
           * @return <code>true</code> if this factory creates <code>writetime</code> selectors instances,
           * <code>false</code> otherwise
           */
          public boolean isWritetimeSelectorFactory()
          {
              return false;
          }
  
          /**
           * Checks if this factory creates <code>maxwritetime</code> selector instances.
           *
           * @return <code>true</code> if this factory creates <code>maxwritetime</code> selectors instances,
           * <code>false</code> otherwise
           */
          public boolean isMaxWritetimeSelectorFactory()
          {
              return false;
          }
  
          /**
           * Checks if this factory creates <code>TTL</code> selectors instances.
           *
           * @return <code>true</code> if this factory creates <code>TTL</code> selectors instances,
           * <code>false</code> otherwise
           */
          public boolean isTTLSelectorFactory()
          {
              return false;
          }
  
          /**
           * Checks if this factory creates <code>Selector</code>s that simply return a column value.
           *
           * @return <code>true</code> if this factory creates <code>Selector</code>s that simply return a column value,
           * <code>false</code> otherwise.
           */
          public boolean isSimpleSelectorFactory()
          {
              return false;
          }
  
          /**
           * Checks if this factory creates <code>Selector</code>s that simply return the specified column.
           *
           * @param index the column index
           * @return <code>true</code> if this factory creates <code>Selector</code>s that simply return
           * the specified column, <code>false</code> otherwise.
           */
          public boolean isSimpleSelectorFactoryFor(int index)
          {
              return false;
          }
  
          /**
           * Returns the name of the column corresponding to the output value of the selector instances created by
           * this factory.
           *
           * @return a column name
           */
          protected abstract String getColumnName();
  
          /**
           * Returns the type of the values returned by the selector instances created by this factory.
           *
           * @return the selector output type
           */
          protected abstract AbstractType<?> getReturnType();
  
          /**
           * Record a mapping between the ColumnDefinitions that are used by the selector
           * instances created by this factory and a column in the ResultSet.Metadata
           * returned with a query. In most cases, this is likely to be a 1:1 mapping,
           * but some selector instances may utilise multiple columns (or none at all)
           * to produce a value (i.e. functions).
           *
           * @param mapping the instance of the column mapping belonging to the current query's Selection
           * @param resultsColumn the column in the ResultSet.Metadata to which the ColumnDefinitions used
           *                      by the Selector are to be mapped
           */
          protected abstract void addColumnMapping(SelectionColumnMapping mapping, ColumnSpecification resultsColumn);
  
          /**
           * Checks if all the columns fetched by the selector created by this factory are known
           * @return {@code true} if all the columns fetched by the selector created by this factory are known,
           * {@code false} otherwise.
           */
          abstract boolean areAllFetchedColumnsKnown();
  
          /**
           * Adds the columns fetched by the selector created by this factory to the provided builder, assuming the
           * factory is terminal (i.e. that {@code isTerminal() == true}).
           *
           * @param builder the column builder to add fetched columns (and potential subselection) to.
           * @throws AssertionError if the method is called on a factory where {@code isTerminal()} returns {@code false}.
           */
          abstract void addFetchedColumns(ColumnFilter.Builder builder);
      }
  
      public static class Serializer
      {
          public void serialize(Selector selector, DataOutputPlus out, int version) throws IOException
          {
              out.writeByte(selector.kind().ordinal());
              selector.serialize(out, version);
          }
  
          public Selector deserialize(DataInputPlus in, int version, TableMetadata metadata) throws IOException
          {
              Kind kind = Kind.values()[in.readUnsignedByte()];
              return kind.deserializer.deserialize(in, version, metadata);
          }
  
          public int serializedSize(Selector selector, int version)
          {
              return TypeSizes.sizeof((byte) selector.kind().ordinal()) + selector.serializedSize(version);
          }
      }
  
      /**
       * The {@code Selector} serializer.
       */
      public static final Serializer serializer = new Serializer();
  
      /**
       * The {@code Selector} kind.
       */
      private final Kind kind;
  
      /**
       * Returns the {@code Selector} kind.
       * @return the {@code Selector} kind
       */
      public final Kind kind()
      {
          return kind;
      }
  
      protected Selector(Kind kind)
      {
          this.kind = kind;
      }
  
      /**
       * Add to the provided builder the column (and potential subselections) to fetch for this
       * selection.
       *
       * @param builder the builder to add columns and subselections to.
       */
      public abstract void addFetchedColumns(ColumnFilter.Builder builder);
  
      /**
       * A row of data that need to be processed by a {@code Selector}
       */
      public static final class InputRow
      {
          private final ProtocolVersion protocolVersion;
          private final List<ColumnMetadata> columns;
          private final boolean unmask;
          private final boolean collectWritetimes;
          private final boolean collectTTLs;
  
          private ByteBuffer[] values;
          private RowTimestamps writetimes;
          private RowTimestamps ttls;
          private int index;
  
          public InputRow(ProtocolVersion protocolVersion, List<ColumnMetadata> columns, boolean unmask)
          {
              this(protocolVersion, columns, unmask, false, false);
          }
  
          public InputRow(ProtocolVersion protocolVersion,
                          List<ColumnMetadata> columns,
                          boolean unmask,
                          boolean collectWritetimes,
                          boolean collectTTLs)
          {
              this.protocolVersion = protocolVersion;
              this.columns = columns;
              this.unmask = unmask;
              this.collectWritetimes = collectWritetimes;
              this.collectTTLs = collectTTLs;
  
              values = new ByteBuffer[columns.size()];
              writetimes = initTimestamps(TimestampsType.WRITETIMES, collectWritetimes, columns);
              ttls = initTimestamps(TimestampsType.TTLS, collectTTLs, columns);
          }
  
          private RowTimestamps initTimestamps(TimestampsType type,
                                               boolean collectWritetimes,
                                               List<ColumnMetadata> columns)
          {
              return collectWritetimes ? RowTimestamps.newInstance(type, columns)
                                       : RowTimestamps.NOOP_ROW_TIMESTAMPS;
          }
  
          public ProtocolVersion getProtocolVersion()
          {
              return protocolVersion;
          }
  
          public boolean unmask()
          {
              return unmask;
          }
  
          public void add(ByteBuffer v)
          {
              values[index] = v;
  
              if (v != null)
              {
                  writetimes.addNoTimestamp(index);
                  ttls.addNoTimestamp(index);
              }
              index++;
          }
  
          public void add(ColumnData columnData, long nowInSec)
          {
              ColumnMetadata column = columns.get(index);
              if (columnData == null)
              {
                  add(null);
              }
              else
              {
                  if (column.isComplex())
                  {
                      add((ComplexColumnData) columnData, nowInSec);
                  }
                  else
                  {
                      add((Cell<?>) columnData, nowInSec);
                  }
              }
          }
  
          private void add(Cell<?> c, long nowInSec)
          {
              values[index] = value(c);
              writetimes.addTimestamp(index, c, nowInSec);
              ttls.addTimestamp(index, c, nowInSec);
              index++;
          }
  
          private void add(ComplexColumnData ccd, long nowInSec)
          {
              AbstractType<?> type = columns.get(index).type;
              if (type.isCollection())
              {
                  values[index] = ((CollectionType<?>) type).serializeForNativeProtocol(ccd.iterator());
  
                  for (Cell<?> cell : ccd)
                  {
                      writetimes.addTimestamp(index, cell, nowInSec);
                      ttls.addTimestamp(index, cell, nowInSec);
                  }
              }
              else
              {
                  UserType udt = (UserType) type;
                  int size = udt.size();
  
                  values[index] = udt.serializeForNativeProtocol(ccd.iterator());
  
                  short fieldPosition = 0;
                  for (Cell<?> cell : ccd)
                  {
                      // handle null fields that aren't at the end
                      short fieldPositionOfCell = ByteBufferUtil.toShort(cell.path().get(0));
                      while (fieldPosition < fieldPositionOfCell)
                      {
                          fieldPosition++;
                          writetimes.addNoTimestamp(index);
                          ttls.addNoTimestamp(index);
                      }
  
                      fieldPosition++;
                      writetimes.addTimestamp(index, cell, nowInSec);
                      ttls.addTimestamp(index, cell, nowInSec);
                  }
  
                  // append nulls for missing cells
                  while (fieldPosition < size)
                  {
                      fieldPosition++;
                      writetimes.addNoTimestamp(index);
                      ttls.addNoTimestamp(index);
                  }
              }
              index++;
          }
  
          private <V> ByteBuffer value(Cell<V> c)
          {
              return c.isCounterCell()
                   ? ByteBufferUtil.bytes(CounterContext.instance().total(c.value(), c.accessor()))
                   : c.buffer();
          }
  
          /**
           * Return the value of the column with the specified index.
           *
           * @param index the column index
           * @return the value of the column with the specified index
           */
          public ByteBuffer getValue(int index)
          {
              return values[index];
          }
  
          /**
           * Reset the row internal state.
           * <p>If the reset is not a deep one only the index will be reset. If the reset is a deep one a new
           * array will be created to store the column values. This allow to reduce object creation when it is not
           * necessary.</p>
           *
           * @param deep {@code true} if the reset must be a deep one.
           */
          public void reset(boolean deep)
          {
              index = 0;
              this.writetimes = initTimestamps(TimestampsType.WRITETIMES, collectWritetimes, columns);
              this.ttls = initTimestamps(TimestampsType.TTLS, collectTTLs, columns);
  
              if (deep)
                  values = new ByteBuffer[values.length];
          }
  
          /**
           * Return the timestamp of the column component with the specified indexes.
           *
           * @return the timestamp of the cell with the specified indexes
           */
          ColumnTimestamps getWritetimes(int columnIndex)
          {
              return writetimes.get(columnIndex);
          }
  
          /**
           * Return the ttl of the column component with the specified column and cell indexes.
           *
           * @param columnIndex the column index
           * @return the ttl of the column with the specified indexes
           */
          ColumnTimestamps getTtls(int columnIndex)
          {
              return ttls.get(columnIndex);
          }
  
          /**
           * Returns the column values as list.
           * <p>This content of the list will be shared with the {@code InputRow} unless a deep reset has been done.</p>
           *
           * @return the column values as list.
           */
          public List<ByteBuffer> getValues()
          {
              return Arrays.asList(values);
          }
      }
  
      /**
       * Add the current value from the specified <code>ResultSetBuilder</code>.
       *
       * @param input the input row
       * @throws InvalidRequestException if a problem occurs while adding the input row
       */
      public abstract void addInput(InputRow input);
  
      /**
       * Returns the selector output.
       *
       * @param protocolVersion protocol version used for serialization
       * @return the selector output
       * @throws InvalidRequestException if a problem occurs while computing the output value
       */
      public abstract ByteBuffer getOutput(ProtocolVersion protocolVersion) throws InvalidRequestException;
  
      protected ColumnTimestamps getWritetimes(ProtocolVersion protocolVersion)
      {
          throw new UnsupportedOperationException();
      }
  
      protected ColumnTimestamps getTTLs(ProtocolVersion protocolVersion)
      {
          throw new UnsupportedOperationException();
      }
  
      /**
       * Returns the <code>Selector</code> output type.
       *
       * @return the <code>Selector</code> output type.
       */
      public abstract AbstractType<?> getType();
  
      /**
       * Reset the internal state of this <code>Selector</code>.
       */
      public abstract void reset();
  
      /**
       * A selector is terminal if it doesn't require any input for it's output to be computed, i.e. if {@link #getOutput}
       * result doesn't depend of {@link #addInput}. This is typically the case of a constant value or functions on constant
       * values.
       */
      public boolean isTerminal()
      {
          return false;
      }
  
      /**
       * Checks that this selector is valid for GROUP BY clause.
       */
      public void validateForGroupBy()
      {
          throw invalidRequest("Only column names and monotonic scalar functions are supported in the GROUP BY clause.");
      }
  
      protected abstract int serializedSize(int version);
  
      protected abstract void serialize(DataOutputPlus out, int version) throws IOException;
  
      protected static void writeType(DataOutputPlus out, AbstractType<?> type) throws IOException
      {
          out.writeUTF(type.asCQL3Type().toString());
      }
  
      protected static int sizeOf(AbstractType<?> type)
      {
          return TypeSizes.sizeof(type.asCQL3Type().toString());
      }
  }
  