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.interpreter;
  
  import org.apache.calcite.DataContext;
  import org.apache.calcite.linq4j.Enumerable;
  import org.apache.calcite.linq4j.Queryable;
  import org.apache.calcite.plan.RelOptTable;
  import org.apache.calcite.plan.RelOptUtil;
  import org.apache.calcite.rel.core.TableScan;
  import org.apache.calcite.rel.type.RelDataType;
  import org.apache.calcite.rel.type.RelDataTypeFactory;
  import org.apache.calcite.rel.type.RelDataTypeField;
  import org.apache.calcite.rex.RexNode;
  import org.apache.calcite.rex.RexUtil;
  import org.apache.calcite.runtime.Enumerables;
  import org.apache.calcite.schema.FilterableTable;
  import org.apache.calcite.schema.ProjectableFilterableTable;
  import org.apache.calcite.schema.QueryableTable;
  import org.apache.calcite.schema.ScannableTable;
  import org.apache.calcite.schema.SchemaPlus;
  import org.apache.calcite.schema.Schemas;
  import org.apache.calcite.util.ImmutableBitSet;
  import org.apache.calcite.util.ImmutableIntList;
  import org.apache.calcite.util.ReflectUtil;
  import org.apache.calcite.util.Util;
  import org.apache.calcite.util.mapping.Mapping;
  import org.apache.calcite.util.mapping.Mappings;
  
  import com.google.common.collect.ImmutableList;
  import com.google.common.collect.Iterables;
  import com.google.common.collect.Lists;
  
  import org.checkerframework.checker.nullness.qual.Nullable;
  
  import java.lang.reflect.Field;
  import java.lang.reflect.Type;
  import java.util.List;
  
  import static org.apache.calcite.util.Static.RESOURCE;
  
  import static java.util.Objects.requireNonNull;
  
  /**
   * Interpreter node that implements a
   * {@link org.apache.calcite.rel.core.TableScan}.
   */
  public class TableScanNode implements Node {
    private TableScanNode(Compiler compiler, TableScan rel,
        Enumerable<Row> enumerable) {
      compiler.enumerable(rel, enumerable);
    }
  
    @Override public void run() {
      // nothing to do
    }
  
    /** Creates a TableScanNode.
     *
     * <p>Tries various table SPIs, and negotiates with the table which filters
     * and projects it can implement. Adds to the Enumerable implementations of
     * any filters and projects that cannot be implemented by the table. */
    static TableScanNode create(Compiler compiler, TableScan rel,
        ImmutableList<RexNode> filters, @Nullable ImmutableIntList projects) {
      final RelOptTable relOptTable = rel.getTable();
      final ProjectableFilterableTable pfTable =
          relOptTable.unwrap(ProjectableFilterableTable.class);
      if (pfTable != null) {
        return createProjectableFilterable(compiler, rel, filters, projects,
            pfTable);
      }
      final FilterableTable filterableTable =
          relOptTable.unwrap(FilterableTable.class);
      if (filterableTable != null) {
        return createFilterable(compiler, rel, filters, projects,
            filterableTable);
      }
      final ScannableTable scannableTable =
          relOptTable.unwrap(ScannableTable.class);
      if (scannableTable != null) {
        return createScannable(compiler, rel, filters, projects,
            scannableTable);
      }
      //noinspection unchecked
      final Enumerable<Row> enumerable = relOptTable.unwrap(Enumerable.class);
      if (enumerable != null) {
        return createEnumerable(compiler, rel, enumerable, null, filters,
            projects);
      }
      final QueryableTable queryableTable =
          relOptTable.unwrap(QueryableTable.class);
      if (queryableTable != null) {
        return createQueryable(compiler, rel, filters, projects,
            queryableTable);
      }
      throw new AssertionError("cannot convert table " + relOptTable
          + " to enumerable");
    }
  
    private static TableScanNode createScannable(Compiler compiler, TableScan rel,
        ImmutableList<RexNode> filters, @Nullable ImmutableIntList projects,
        ScannableTable scannableTable) {
      final Enumerable<Row> rowEnumerable =
          Enumerables.toRow(scannableTable.scan(compiler.getDataContext()));
      return createEnumerable(compiler, rel, rowEnumerable, null, filters,
          projects);
    }
  
    private static TableScanNode createQueryable(Compiler compiler,
        TableScan rel, ImmutableList<RexNode> filters, @Nullable ImmutableIntList projects,
        QueryableTable queryableTable) {
      final DataContext root = compiler.getDataContext();
      final RelOptTable relOptTable = rel.getTable();
      final Type elementType = queryableTable.getElementType();
      SchemaPlus schema = root.getRootSchema();
      for (String name : Util.skipLast(relOptTable.getQualifiedName())) {
        requireNonNull(schema, () ->
            "schema is null while resolving " + name + " for table"
                + relOptTable.getQualifiedName());
        schema = schema.subSchemas().get(name);
      }
      final Enumerable<Row> rowEnumerable;
      if (elementType instanceof Class) {
        //noinspection unchecked
        final Queryable<Object> queryable =
            Schemas.queryable(root, (Class) elementType,
                relOptTable.getQualifiedName());
        ImmutableList.Builder<Field> fieldBuilder = ImmutableList.builder();
        Class type = (Class) elementType;
        for (Field field : type.getFields()) {
          if (ReflectUtil.isPublic(field) && !ReflectUtil.isStatic(field)) {
            fieldBuilder.add(field);
          }
        }
        final List<Field> fields = fieldBuilder.build();
        rowEnumerable = queryable.select(o -> {
          final @Nullable Object[] values = new Object[fields.size()];
          for (int i = 0; i < fields.size(); i++) {
            Field field = fields.get(i);
            try {
              values[i] = field.get(o);
            } catch (IllegalAccessException e) {
              throw new RuntimeException(e);
            }
          }
          return new Row(values);
        });
      } else {
        rowEnumerable =
            Schemas.queryable(root, Row.class, relOptTable.getQualifiedName());
      }
      return createEnumerable(compiler, rel, rowEnumerable, null, filters,
          projects);
    }
  
    private static TableScanNode createFilterable(Compiler compiler,
        TableScan rel, ImmutableList<RexNode> filters, @Nullable ImmutableIntList projects,
        FilterableTable filterableTable) {
      final DataContext root = compiler.getDataContext();
      final List<RexNode> mutableFilters = Lists.newArrayList(filters);
      final Enumerable<@Nullable Object[]> enumerable =
          filterableTable.scan(root, mutableFilters);
      for (RexNode filter : mutableFilters) {
        if (!filters.contains(filter)) {
          throw RESOURCE.filterableTableInventedFilter(filter.toString()).ex();
        }
      }
      final Enumerable<Row> rowEnumerable = Enumerables.toRow(enumerable);
      return createEnumerable(compiler, rel, rowEnumerable, null,
          mutableFilters, projects);
    }
  
    private static TableScanNode createProjectableFilterable(Compiler compiler,
        TableScan rel, ImmutableList<RexNode> filters, @Nullable ImmutableIntList projects,
        ProjectableFilterableTable pfTable) {
      final DataContext root = compiler.getDataContext();
      final ImmutableIntList originalProjects = projects;
      for (;;) {
        final List<RexNode> mutableFilters = Lists.newArrayList(filters);
        final int[] projectInts;
        if (projects == null
            || projects.equals(TableScan.identity(rel.getTable()))) {
          projectInts = null;
        } else {
          projectInts = projects.toIntArray();
        }
        for (RexNode filter : mutableFilters) {
          if (!filters.contains(filter)) {
            throw RESOURCE.filterableTableInventedFilter(filter.toString())
                .ex();
          }
        }
        final ImmutableBitSet usedFields =
            RelOptUtil.InputFinder.bits(mutableFilters, null);
        if (projects != null) {
          int changeCount = 0;
          for (int usedField : usedFields) {
            if (!projects.contains(usedField)) {
              // A field that is not projected is used in a filter that the
              // table rejected. We won't be able to apply the filter later.
              // Try again without any projects.
              projects =
                  ImmutableIntList.copyOf(
                      Iterables.concat(projects, ImmutableList.of(usedField)));
              ++changeCount;
            }
          }
          if (changeCount > 0) {
            continue;
          }
        }
        final Enumerable<@Nullable Object[]> enumerable1 =
            pfTable.scan(root, mutableFilters, projectInts);
        final Enumerable<Row> rowEnumerable = Enumerables.toRow(enumerable1);
        final ImmutableIntList rejectedProjects;
        if (originalProjects == null || originalProjects.equals(projects)) {
          rejectedProjects = null;
        } else {
          // We projected extra columns because they were needed in filters. Now
          // project the leading columns.
          rejectedProjects = ImmutableIntList.identity(originalProjects.size());
        }
        return createEnumerable(compiler, rel, rowEnumerable, projects,
            mutableFilters, rejectedProjects);
      }
    }
  
    private static TableScanNode createEnumerable(Compiler compiler,
        TableScan rel, Enumerable<Row> enumerable,
        final @Nullable ImmutableIntList acceptedProjects, List<RexNode> rejectedFilters,
        final @Nullable ImmutableIntList rejectedProjects) {
      if (!rejectedFilters.isEmpty()) {
        final RexNode filter =
            RexUtil.composeConjunction(rel.getCluster().getRexBuilder(),
                rejectedFilters);
        // Re-map filter for the projects that have been applied already
        final RexNode filter2;
        final RelDataType inputRowType;
        if (acceptedProjects == null) {
          filter2 = filter;
          inputRowType = rel.getRowType();
        } else {
          final Mapping mapping =
              Mappings.target(acceptedProjects,
                  rel.getTable().getRowType().getFieldCount());
          filter2 = RexUtil.apply(mapping, filter);
          final RelDataTypeFactory.Builder builder =
              rel.getCluster().getTypeFactory().builder();
          final List<RelDataTypeField> fieldList =
              rel.getTable().getRowType().getFieldList();
          for (int acceptedProject : acceptedProjects) {
            builder.add(fieldList.get(acceptedProject));
          }
          inputRowType = builder.build();
        }
        final Scalar condition =
            compiler.compile(ImmutableList.of(filter2), inputRowType);
        final Context context = compiler.createContext();
        enumerable = enumerable.where(row -> {
          context.values = row.getValues();
          Boolean b = (Boolean) condition.execute(context);
          return b != null && b;
        });
      }
      if (rejectedProjects != null) {
        final @Nullable Object[] values = new Object[rejectedProjects.size()];
        enumerable =
            enumerable.select(row -> {
              final @Nullable Object[] inValues = row.getValues();
              for (int i = 0; i < rejectedProjects.size(); i++) {
                values[i] = inValues[rejectedProjects.get(i)];
              }
              return Row.asCopy(values);
            });
      }
      return new TableScanNode(compiler, rel, enumerable);
    }
  }
  