@Getter
public class SqlMetadata {
    private static final Logger LOGGER = LoggerFactory.getLogger(SqlMetadata.class);
    private final GraphTraversalSource g;
    private final GremlinSchema gremlinSchema;
    private final Map<String, String> tableRenameMap = new HashMap<>();
    private final Map<String, String> columnRenameMap = new HashMap<>();
    private final Map<String, List<String>> columnOutputListMap = new HashMap<>();
    private boolean isAggregate = false;

    public SqlMetadata(final GraphTraversalSource g, final GremlinSchema gremlinSchema) {
        this.g = g;
        this.gremlinSchema = gremlinSchema;
    }

    private static boolean isAggregate(final SqlNode sqlNode) {
        if (sqlNode instanceof SqlCall) {
            final SqlCall sqlCall = (SqlCall) sqlNode;
            if (isAggregate(sqlCall.getOperator())) {
                return true;
            }
            for (final SqlNode tmpSqlNode : sqlCall.getOperandList()) {
                if (isAggregate(tmpSqlNode)) {
                    return true;
                }
            }
        }
        return false;
    }

    private static boolean isAggregate(final SqlOperator sqlOperator) {
        return sqlOperator instanceof SqlAggFunction;
    }

    public boolean getIsColumnEdge(final String tableName, final String columnName) throws SQLException {
        return getGremlinTable(tableName).getIsVertex() &&
                (columnName.endsWith(GremlinTableBase.IN_ID) || columnName.endsWith(GremlinTableBase.OUT_ID));
    }

    public String getColumnEdgeLabel(final String column) throws SQLException {
        final String columnName = getRenamedColumn(column);
        final GremlinTableBase gremlinTableBase;
        if (columnName.endsWith(GremlinTableBase.IN_ID)) {
            gremlinTableBase = getGremlinTable(column.substring(0, column.length() - GremlinTableBase.IN_ID.length()));
        } else if (columnName.endsWith(GremlinTableBase.OUT_ID)) {
            gremlinTableBase = getGremlinTable(column.substring(0, column.length() - GremlinTableBase.OUT_ID.length()));
        } else {
            throw new SQLException(String.format("Error: Edge labels must end with %s or %s.", GremlinTableBase.IN_ID,
                    GremlinTableBase.OUT_ID));
        }

        if (gremlinTableBase.getIsVertex()) {
            throw new SQLException("Error: Expected edge table.");
        }
        return gremlinTableBase.getLabel();
    }

    public boolean isLeftInRightOut(final String leftVertexLabel, final String rightVertexLabel) {
        for (final GremlinVertexTable gremlinVertexTable : gremlinSchema.getVertices()) {
            if (gremlinVertexTable.hasInEdge(leftVertexLabel) && gremlinVertexTable.hasOutEdge(rightVertexLabel)) {
                return true;
            }
        }
        return false;
    }

    public boolean isRightInLeftOut(final String leftVertexLabel, final String rightVertexLabel) {
        for (final GremlinVertexTable gremlinVertexTable : gremlinSchema.getVertices()) {
            if (gremlinVertexTable.hasInEdge(rightVertexLabel) && gremlinVertexTable.hasOutEdge(leftVertexLabel)) {
                return true;
            }
        }
        return false;
    }

    public Set<String> getRenamedColumns() {
        return new HashSet<>(columnRenameMap.keySet());
    }

    public void setColumnOutputList(final String table, final List<String> columnOutputList) {
        columnOutputListMap.put(table, new ArrayList<>(columnOutputList));
    }

    public Set<GremlinTableBase> getTables() throws SQLException {
        final Set<GremlinTableBase> tables = new HashSet<>();
        for (final String table : tableRenameMap.values()) {
            tables.add(getGremlinTable(table));
        }
        return tables;
    }

    public boolean isVertex(final String table) throws SQLException {
        final String renamedTableName = getRenamedTable(table);
        for (final GremlinVertexTable gremlinVertexTable : gremlinSchema.getVertices()) {
            if (gremlinVertexTable.getLabel().equalsIgnoreCase(renamedTableName)) {
                return true;
            }
        }
        for (final GremlinEdgeTable gremlinEdgeTable : gremlinSchema.getEdges()) {
            if (gremlinEdgeTable.getLabel().equalsIgnoreCase(renamedTableName)) {
                return false;
            }
        }
        throw new SQLException("Error: Table {} does not exist.", renamedTableName);
    }

    public GremlinTableBase getGremlinTable(final String table) throws SQLException {
        final String renamedTableName = getRenamedTable(table);
        for (final GremlinTableBase gremlinTableBase : gremlinSchema.getAllTables()) {
            if (gremlinTableBase.getLabel().equalsIgnoreCase(renamedTableName)) {
                return gremlinTableBase;
            }
        }
        throw new SQLException(String.format("Error: Table %s does not exist.", renamedTableName));
    }

    public void addRenamedTable(final String actualName, final String renameName) {
        tableRenameMap.put(renameName, actualName);
    }

    public String getRenamedTable(final String table) {
        return tableRenameMap.getOrDefault(table, table);
    }

    public void addRenamedColumn(final String actualName, final String renameName) {
        columnRenameMap.put(renameName, actualName);
    }

    public String getRenamedColumn(final String column) {
        return columnRenameMap.getOrDefault(column, column);
    }

    public String getRenameFromActual(final String actual) {
        final Optional<Map.Entry<String, String>>
                rename = tableRenameMap.entrySet().stream().filter(t -> t.getValue().equals(actual)).findFirst();
        if (rename.isPresent()) {
            return rename.get().getKey();
        }
        return actual;
    }

    public String getActualColumnName(final GremlinTableBase table, final String column) throws SQLException {
        final String actualColumnName = getRenamedColumn(column);
        for (final GremlinProperty gremlinProperty : table.getColumns().values()) {
            if (gremlinProperty.getName().equalsIgnoreCase(actualColumnName)) {
                return gremlinProperty.getName();
            }
        }
        throw new SQLException(
                String.format("Error: Column %s does not exist in table %s.", actualColumnName, table.getLabel()));
    }

    public boolean getTableHasColumn(final GremlinTableBase table, final String column) {
        final String actualColumnName = getRenamedColumn(column);
        for (final GremlinProperty gremlinProperty : table.getColumns().values()) {
            if (gremlinProperty.getName().equalsIgnoreCase(actualColumnName)) {
                return true;
            }
        }
        return false;
    }

    public String getActualTableName(final String table) throws SQLException {
        final String renamedTableName = getRenamedTable(table);
        for (final GremlinVertexTable gremlinVertexTable : gremlinSchema.getVertices()) {
            if (gremlinVertexTable.getLabel().equalsIgnoreCase(renamedTableName)) {
                return gremlinVertexTable.getLabel();
            }
        }
        for (final GremlinEdgeTable gremlinEdgeTable : gremlinSchema.getEdges()) {
            if (gremlinEdgeTable.getLabel().equalsIgnoreCase(renamedTableName)) {
                return gremlinEdgeTable.getLabel();
            }
        }
        throw new SQLException(String.format("Error: Table %s.", table));
    }

    public void checkAggregate(final SqlNodeList sqlNodeList) {
        isAggregate = sqlNodeList.getList().stream().allMatch(SqlMetadata::isAggregate);
    }

    public boolean getIsAggregate() {
        return isAggregate;
    }

    public GremlinProperty getGremlinProperty(final String table, final String column) throws SQLException {
        final String actualColumnName = getActualColumnName(getGremlinTable(table), column);
        return getGremlinTable(table).getColumn(actualColumnName);
    }
} public class SqlTraversalEngine {
    private static final Logger LOGGER = LoggerFactory.getLogger(SqlTraversalEngine.class);

    public static GraphTraversal<?, ?> generateInitialSql(final List<GremlinSqlIdentifier> gremlinSqlIdentifiers,
                                                          final SqlMetadata sqlMetadata,
                                                          final GraphTraversalSource g) throws SQLException {
        if (gremlinSqlIdentifiers.size() != 2) {
            throw new SQLException("Expected GremlinSqlIdentifier List size to be 2.");
        }
        final String label = sqlMetadata.getActualTableName(gremlinSqlIdentifiers.get(0).getName(1));
        final GraphTraversal<?, ?> graphTraversal = sqlMetadata.isVertex(label) ? g.V() : g.E();
        graphTraversal.hasLabel(label);
        return graphTraversal;
    }

    public static void applyAggregateFold(final SqlMetadata sqlMetadata, final GraphTraversal<?, ?> graphTraversal) {
        if (sqlMetadata.getIsAggregate()) {
            graphTraversal.fold();
        }
    }

    public static GraphTraversal<?, ?> getEmptyTraversal(final StepDirection direction, final SqlMetadata sqlMetadata) {
        final GraphTraversal<?, ?> graphTraversal = __.unfold();
        if (sqlMetadata.getIsAggregate()) {
            graphTraversal.unfold();
        }
        switch (direction) {
            case Out:
                return graphTraversal.outV();
            case In:
                return graphTraversal.inV();
        }
        return graphTraversal;
    }

    public static void addProjection(final List<GremlinSqlIdentifier> gremlinSqlIdentifiers,
                                     final SqlMetadata sqlMetadata,
                                     final GraphTraversal<?, ?> graphTraversal) throws SQLException {
        if (gremlinSqlIdentifiers.size() != 2) {
            throw new SQLException("Expected GremlinSqlIdentifier List size to be 2.");
        }
        final String label = sqlMetadata.getActualTableName(gremlinSqlIdentifiers.get(0).getName(1));
        final String projectLabel = gremlinSqlIdentifiers.get(1).getName(0);

        graphTraversal.project(projectLabel);
        sqlMetadata.addRenamedTable(label, projectLabel);
    }

    public static GraphTraversal<?, ?> getEmptyTraversal(final SqlMetadata sqlMetadata) {
        return getEmptyTraversal(StepDirection.None, sqlMetadata);
    }


    public static void applyTraversal(final GraphTraversal<?, ?> graphTraversal,
                                      final GraphTraversal<?, ?> subGraphTraversal) {
        graphTraversal.by(subGraphTraversal);
    }

    public static void applySqlIdentifier(final GremlinSqlIdentifier sqlIdentifier,
                                          final SqlMetadata sqlMetadata,
                                          final GraphTraversal<?, ?> graphTraversal) throws SQLException {
        // Format of identifier is 'table'.'column => ['table', 'column']
        appendGraphTraversal(sqlIdentifier.getName(0), sqlIdentifier.getName(1), sqlMetadata, graphTraversal);
    }

    public static GraphTraversal<?, ?> applyColumnRenames(final List<String> columnsRenamed) throws SQLException {
        final String firstColumn = columnsRenamed.remove(0);
        final String[] remaining = columnsRenamed.toArray(new String[] {});
        return __.project(firstColumn, remaining);
    }

    private static void appendGraphTraversal(final String table, final String column,
                                             final SqlMetadata sqlMetadata,
                                             final GraphTraversal<?, ?> graphTraversal) throws SQLException {
        final GremlinTableBase gremlinTableBase = sqlMetadata.getGremlinTable(table);
        final String columnName = sqlMetadata.getActualColumnName(gremlinTableBase, column);

        // Primary/foreign key, need to traverse appropriately.
        if (!columnName.endsWith(GremlinTableBase.ID)) {
            if (sqlMetadata.getIsAggregate()) {
                graphTraversal.values(columnName);
            } else {
                graphTraversal.choose(__.has(columnName), __.values(columnName),
                        __.constant(SqlGremlinQueryResult.NULL_VALUE));
            }
        } else {
            // It's this vertex/edge.
            if (columnName.toLowerCase().startsWith(gremlinTableBase.getLabel())) {
                graphTraversal.id();
            } else {
                if (columnName.endsWith(IN_ID)) {
                    // Vertices can have many connected, edges (thus we need to fold). Edges can only connect to 1 vertex.
                    if (gremlinTableBase.getIsVertex()) {
                        graphTraversal.coalesce(__.inE().hasLabel(columnName.replace(IN_ID, "")).id().fold(),
                                __.constant(new ArrayList<>()));
                    } else {
                        graphTraversal.coalesce(__.inV().hasLabel(columnName.replace(IN_ID, "")).id(),
                                __.constant(new ArrayList<>()));
                    }
                } else if (column.endsWith(OUT_ID)) {
                    // Vertices can have many connected, edges (thus we need to fold). Edges can only connect to 1 vertex.
                    if (gremlinTableBase.getIsVertex()) {
                        graphTraversal.coalesce(__.outE().hasLabel(columnName.replace(OUT_ID, "")).id().fold(),
                                __.constant(new ArrayList<>()));
                    } else {
                        graphTraversal.coalesce(__.outV().hasLabel(columnName.replace(IN_ID, "")).id(),
                                __.constant(new ArrayList<>()));
                    }
                } else {
                    graphTraversal.constant(new ArrayList<>());
                }
            }
        }
    }
} public class GremlinSqlFactory {
    private static SqlMetadata sqlMetadata = null;

    public static void setSqlMetadata(final SqlMetadata sqlMetadata1) {
        sqlMetadata = sqlMetadata1;
    }

    public static SqlMetadata getGremlinSqlMetadata() throws SQLException {
        if (sqlMetadata == null) {
            throw new SQLException("Error: Schema must be set.");
        }
        return sqlMetadata;
    }

    public static GremlinSqlJoinComparison createJoinEquality(final SqlNode sqlNode)
            throws SQLException {
        if (sqlNode instanceof SqlBasicCall) {
            final SqlBasicCall sqlBasicCall = (SqlBasicCall) sqlNode;
            if (sqlBasicCall.getOperator() instanceof SqlBinaryOperator) {
                return new GremlinSqlJoinComparison((SqlBasicCall) sqlNode,
                        (SqlBinaryOperator) sqlBasicCall.getOperator(), createNodeList(sqlBasicCall.getOperandList()),
                        getGremlinSqlMetadata());
            }
        }
        throw new SQLException(String.format("Error: Unknown node: %s.", sqlNode.getClass().getName()));
    }

    public static GremlinSqlOperator createOperator(final SqlOperator sqlOperator, final List<SqlNode> sqlOperands)
            throws SQLException {
        if (sqlOperator instanceof SqlAsOperator) {
            return new GremlinSqlAsOperator((SqlAsOperator) sqlOperator, createNodeList(sqlOperands),
                    getGremlinSqlMetadata());
        } else if (sqlOperator instanceof SqlAggFunction) {
            return new GremlinSqlAggFunction((SqlAggFunction) sqlOperator, createNodeList(sqlOperands),
                    getGremlinSqlMetadata());
        } else if (sqlOperator instanceof SqlBinaryOperator) {
            return new GremlinSqlBinaryOperator((SqlBinaryOperator) sqlOperator, createNodeList(sqlOperands),
                    getGremlinSqlMetadata());
        } else if (sqlOperator instanceof SqlPostfixOperator) {
            return new GremlinSqlPostFixOperator((SqlPostfixOperator) sqlOperator, createNodeList(sqlOperands),
                    getGremlinSqlMetadata());
        }
        throw new SQLException(String.format("Error: Unknown operator: %s.", sqlOperator.getKind().sql));
    }

    public static GremlinSqlNode createNode(final SqlNode sqlNode) throws SQLException {
        if (sqlNode instanceof SqlBasicCall) {
            return new GremlinSqlBasicCall((SqlBasicCall) sqlNode, getGremlinSqlMetadata());
        } else if (sqlNode instanceof SqlIdentifier) {
            return new GremlinSqlIdentifier((SqlIdentifier) sqlNode, getGremlinSqlMetadata());
        } else if (sqlNode instanceof SqlNumericLiteral) {
            return new GremlinSqlNumericLiteral((SqlNumericLiteral) sqlNode, getGremlinSqlMetadata());
        }
        throw new SQLException(String.format("Error: Unknown node: %s.", sqlNode.getClass().getName()));
    }

    public static List<GremlinSqlNode> createNodeList(final List<SqlNode> sqlNodes) throws SQLException {
        final List<GremlinSqlNode> gremlinSqlNodes = new ArrayList<>();
        for (final SqlNode sqlNode : sqlNodes) {
            gremlinSqlNodes.add(createNode(sqlNode));
        }
        return gremlinSqlNodes;
    }

    @SuppressWarnings("unchecked")
    public static <T> T createNodeCheckType(final SqlNode sqlNode, final Class<T> clazz) throws SQLException {
        final GremlinSqlNode gremlinSqlNode = createNode(sqlNode);
        if (!gremlinSqlNode.getClass().equals(clazz)) {
            throw new SQLException("Error: Type mismatch.");
        }
        return (T) gremlinSqlNode;
    }

    public static GremlinSqlSelect createSelect(final SqlSelect selectRoot, final GraphTraversalSource g)
            throws SQLException {
        if (selectRoot.getFrom() instanceof SqlJoin) {
            return new GremlinSqlSelectMulti(selectRoot, (SqlJoin) selectRoot.getFrom(), sqlMetadata, g);
        } else if (selectRoot.getFrom() instanceof SqlBasicCall) {
            return new GremlinSqlSelectSingle(selectRoot, (SqlBasicCall) selectRoot.getFrom(), sqlMetadata, g);
        }
        throw new SQLException(String.format("Error: Unknown node for getFrom: %s.", selectRoot.getFrom().getClass().getName()));
    }

    public static boolean isTable(final SqlNode sqlNode, final String renamedTable) throws SQLException {
        if (sqlNode instanceof SqlIdentifier) {
            return (((SqlIdentifier) sqlNode).names.get(0).equalsIgnoreCase(renamedTable));
        } else if (sqlNode instanceof SqlCall) {
            for (final SqlNode tmpSqlNode : ((SqlCall) sqlNode).getOperandList()) {
                if (isTable(tmpSqlNode, renamedTable)) {
                    return true;
                }
            }
        } else {
            throw new SQLException("Error: Unknown node type for isTable.");
        }
        return false;
    }
} @AllArgsConstructor
public abstract class GremlinSqlNode {
    private static final Logger LOGGER = LoggerFactory.getLogger(GremlinSqlNode.class);
    private final SqlNode sqlNode;
    private final SqlMetadata sqlMetadata;
} public class GremlinSqlIdentifier extends GremlinSqlNode {
    private final SqlIdentifier sqlIdentifier;

    public GremlinSqlIdentifier(final SqlIdentifier sqlIdentifier, final SqlMetadata sqlMetadata) {
        super(sqlIdentifier, sqlMetadata);
        this.sqlIdentifier = sqlIdentifier;
    }


    public String getName(final int idx) throws SQLException {
        if (idx >= sqlIdentifier.names.size()) {
            throw new SQLException("Index of identifier > size of name list for identifier");
        }
        return sqlIdentifier.names.get(idx);
    }


    public String getColumn() throws SQLException {
        if (sqlIdentifier.names.size() < 1) {
            throw new SQLException("Expected at least one name in list for identifier");
        }
        return sqlIdentifier.names.get(sqlIdentifier.names.size() - 1);
    }
} public class GremlinSqlAsOperator extends GremlinSqlOperator {
    private static final Logger LOGGER = LoggerFactory.getLogger(GremlinSqlAsOperator.class);
    private final SqlAsOperator sqlAsOperator;
    private final SqlMetadata sqlMetadata;
    private final List<GremlinSqlNode> sqlOperands;

    public GremlinSqlAsOperator(final SqlAsOperator sqlAsOperator, final List<GremlinSqlNode> gremlinSqlNodes,
                                final SqlMetadata sqlMetadata) {
        super(sqlAsOperator, gremlinSqlNodes, sqlMetadata);
        this.sqlAsOperator = sqlAsOperator;
        this.sqlMetadata = sqlMetadata;
        this.sqlOperands = gremlinSqlNodes;
    }

    @Override
    protected void appendTraversal(final GraphTraversal<?, ?> graphTraversal) throws SQLException {
        if (sqlOperands.get(0) instanceof GremlinSqlBasicCall) {
            ((GremlinSqlBasicCall) sqlOperands.get(0)).generateTraversal(graphTraversal);
        } else if (!(sqlOperands.get(0) instanceof GremlinSqlIdentifier) &&
                !(sqlOperands.get(0) instanceof GremlinSqlNumericLiteral)) {
            throw new SQLException(
                    "Error: expected operand to be GremlinSqlBasicCall or GremlinSqlIdentifier in GremlinSqlOperator.");
        }

        if (sqlOperands.size() == 1) {
            if (sqlOperands.get(0) instanceof GremlinSqlIdentifier) {
                SqlTraversalEngine
                        .applySqlIdentifier((GremlinSqlIdentifier) sqlOperands.get(0), sqlMetadata, graphTraversal);
            }
        }
        if (sqlOperands.size() == 2 && sqlOperands.get(0) instanceof GremlinSqlIdentifier) {
            SqlTraversalEngine
                    .applySqlIdentifier((GremlinSqlIdentifier) sqlOperands.get(0), sqlMetadata, graphTraversal);
        }
        sqlMetadata.addRenamedColumn(getActual(), getRename());
    }

    public String getName(final int operandIdx, final int nameIdx) throws SQLException {
        if (operandIdx >= sqlOperands.size() || !(sqlOperands.get(operandIdx) instanceof GremlinSqlIdentifier)) {
            throw new SQLException(
                    "Error: Expected operand idx less than number of operands and GremlinSqlIdentifier for operand");
        }
        return ((GremlinSqlIdentifier) sqlOperands.get(operandIdx)).getName(nameIdx);
    }

    public String getActual() throws SQLException {
        if (sqlOperands.size() != 2) {
            throw new SQLException("Error: Expected two operands for SQL AS statement.");
        }
        if (sqlOperands.get(0) instanceof GremlinSqlIdentifier) {
            return ((GremlinSqlIdentifier) sqlOperands.get(0)).getColumn();
        } else if (sqlOperands.get(0) instanceof GremlinSqlBasicCall) {
            return ((GremlinSqlBasicCall) sqlOperands.get(0)).getActual();
        }
        throw new SQLException("Error, unable to get actual name in GremlinSqlAsOperator.");
    }

    public String getRename() throws SQLException {
        if (sqlOperands.size() != 2) {
            throw new SQLException("Error: Expected two operands for SQL AS statement.");
        }
        if (sqlOperands.get(1) instanceof GremlinSqlIdentifier) {
            return ((GremlinSqlIdentifier) sqlOperands.get(1)).getColumn();
        } else if (sqlOperands.get(1) instanceof GremlinSqlBasicCall) {
            return ((GremlinSqlBasicCall) sqlOperands.get(1)).getRename();
        }
        throw new SQLException("Error, unable to get rename name in GremlinSqlAsOperator.");
    }
} @Getter
public class GremlinSqlBasicCall extends GremlinSqlNode {
    private static final Logger LOGGER = LoggerFactory.getLogger(GremlinSqlBasicCall.class);
    private final SqlBasicCall sqlBasicCall;
    private final GremlinSqlOperator gremlinSqlOperator;
    private final List<GremlinSqlNode> gremlinSqlNodes;

    public GremlinSqlBasicCall(final SqlBasicCall sqlBasicCall, final SqlMetadata sqlMetadata)
            throws SQLException {
        super(sqlBasicCall, sqlMetadata);
        this.sqlBasicCall = sqlBasicCall;
        gremlinSqlOperator =
                GremlinSqlFactory.createOperator(sqlBasicCall.getOperator(), sqlBasicCall.getOperandList());
        gremlinSqlNodes = GremlinSqlFactory.createNodeList(sqlBasicCall.getOperandList());
    }

    void validate() throws SQLException {
        if (gremlinSqlOperator instanceof GremlinSqlAsOperator) {
            if (gremlinSqlNodes.size() != 2) {
                throw new SQLException("Error, expected only two sub nodes for GremlinSqlBasicCall.");
            }
        } else if (gremlinSqlOperator instanceof GremlinSqlAggFunction) {
            if (gremlinSqlNodes.size() != 1) {
                throw new SQLException("Error, expected only one sub node for GremlinSqlAggFunction.");
            }
        }
    }

    public void generateTraversal(final GraphTraversal<?, ?> graphTraversal) throws SQLException {
        validate();
        gremlinSqlOperator.appendOperatorTraversal(graphTraversal);
    }

    public String getRename() throws SQLException {
        if (gremlinSqlOperator instanceof GremlinSqlAsOperator) {
            return ((GremlinSqlAsOperator) gremlinSqlOperator).getRename();
        } else if (gremlinSqlOperator instanceof GremlinSqlAggFunction) {
            if (gremlinSqlNodes.size() == 1 && gremlinSqlNodes.get(0) instanceof GremlinSqlIdentifier) {
                // returns the formatted column name for aggregations
                return ((GremlinSqlAggFunction) gremlinSqlOperator).getNewName();
            }
        }
        throw new SQLException("Unable to determine column rename.");
    }

    public String getActual() throws SQLException {
        if (gremlinSqlOperator instanceof GremlinSqlAsOperator) {
            return ((GremlinSqlAsOperator) gremlinSqlOperator).getActual();
        } else if (gremlinSqlOperator instanceof GremlinSqlAggFunction) {
            if (gremlinSqlNodes.size() == 1 && gremlinSqlNodes.get(0) instanceof GremlinSqlIdentifier) {
                return ((GremlinSqlIdentifier) gremlinSqlNodes.get(0)).getColumn();
            } else if (gremlinSqlNodes.size() == 2 && gremlinSqlNodes.get(1) instanceof GremlinSqlIdentifier) {
                return ((GremlinSqlIdentifier) gremlinSqlNodes.get(1)).getColumn();
            } else if (gremlinSqlNodes.size() == 1 && gremlinSqlNodes.get(0) instanceof GremlinSqlNumericLiteral) {
                return ((GremlinSqlAggFunction) gremlinSqlOperator).getNewName();
            }
        }
        throw new SQLException("Unable to determine actual column name.");
    }
} public class GremlinSqlJoinComparison {
    private static final Logger LOGGER = LoggerFactory.getLogger(GremlinSqlJoinComparison.class);
    // See SqlKind.BINARY_COMPARISON for list of aggregate functions in Calcite.

    private final SqlBasicCall sqlBasicCall;
    private final SqlBinaryOperator sqlBinaryOperator;
    private final SqlMetadata sqlMetadata;
    private final List<GremlinSqlNode> gremlinSqlNodes;


    public GremlinSqlJoinComparison(final SqlBasicCall sqlBasicCall,
                                    final SqlBinaryOperator sqlBinaryOperator,
                                    final List<GremlinSqlNode> gremlinSqlNodes,
                                    final SqlMetadata sqlMetadata) {
        this.sqlBasicCall = sqlBasicCall;
        this.sqlBinaryOperator = sqlBinaryOperator;
        this.sqlMetadata = sqlMetadata;
        this.gremlinSqlNodes = gremlinSqlNodes;
    }

    public boolean isEquals() {
        return sqlBinaryOperator.kind.sql.equals(SqlKind.EQUALS.sql);
    }

    public String getColumn(final String renamedTable) throws SQLException {
        for (final GremlinSqlNode gremlinSqlNode : gremlinSqlNodes) {
            if (!(gremlinSqlNode instanceof GremlinSqlIdentifier)) {
                throw new SQLException("Error: Expected nodes in join comparison to be GremlinSqlIdentifiers.");
            }
            final GremlinSqlIdentifier gremlinSqlIdentifier = (GremlinSqlIdentifier) gremlinSqlNode;
            if (gremlinSqlIdentifier.getName(0).equals(renamedTable)) {
                return gremlinSqlIdentifier.getName(1);
            }
        }
        throw new SQLException("Error: Expected to find join column for renamed table.");
    }
} @Getter
public class GremlinTableBase extends AbstractQueryableTable implements TranslatableTable {
    private final String label;
    private final Boolean isVertex;
    private final Map<String, GremlinProperty> columns;
    public static final String ID = "_ID";
    public static final String IN_ID = "_IN" + ID;
    public static final String OUT_ID = "_OUT" + ID;

    public GremlinTableBase(final String label, final Boolean isVertex,
                            final Map<String, GremlinProperty> columns) {
        super(Object[].class);
        this.label = label;
        this.isVertex = isVertex;
        this.columns = columns;
    }

    public GremlinProperty getColumn(final String column) throws SQLException {
        for (final String key : columns.keySet()) {
            if (key.equalsIgnoreCase(column)) {
                return columns.get(key);
            }
        }
        throw new SQLException(String.format(
                "Error: Could not find column '%s' on %s with label '%s'.", column, isVertex ? "vertex" : "edge", label));
    }

    @Override
    public <T> Queryable<T> asQueryable(final QueryProvider queryProvider, final SchemaPlus schema, final String tableName) {
        return null;
    }

    @Override
    public RelNode toRel(final RelOptTable.ToRelContext context, final RelOptTable relOptTable) {
        final int[] fields = new int[columns.size()];
        for (int i = 0; i < fields.length; i++) {
            fields[i] = i;
        }
        return new GremlinTableScan(context.getCluster(), context.getCluster().traitSetOf(GremlinRel.CONVENTION), relOptTable, fields);
    }

    @Override
    public RelDataType getRowType(final RelDataTypeFactory relDataTypeFactory) {
        final List<String> names = new ArrayList<>();
        final List<RelDataType> types = new ArrayList<>();
        for (final Map.Entry<String, GremlinProperty> entry : columns.entrySet()) {
            names.add(entry.getKey());
            types.add(relDataTypeFactory.createJavaType(getType(entry.getValue().getType())));
        }
        return relDataTypeFactory.createStructType(Pair.zip(names, types));
    }

    private Class<?> getType(final String className) {
        if ("string".equalsIgnoreCase(className)) {
            return String.class;
        } else if ("integer".equalsIgnoreCase(className)) {
            return Integer.class;
        } else if ("float".equalsIgnoreCase(className)) {
            return Float.class;
        } else if ("byte".equalsIgnoreCase(className)) {
            return Byte.class;
        } else if ("short".equalsIgnoreCase(className)) {
            return Short.class;
        } else if ("double".equalsIgnoreCase(className)) {
            return Double.class;
        } else if ("long".equalsIgnoreCase(className)) {
            return Long.class;
        } else if ("boolean".equalsIgnoreCase(className)) {
            return Boolean.class;
        } else if ("date".equalsIgnoreCase(className) || "long_date".equalsIgnoreCase(className)) {
            return java.sql.Date.class;
        } else if ("timestamp".equalsIgnoreCase(className) || "long_timestamp".equalsIgnoreCase(className)) {
            return java.sql.Timestamp.class;
        } else {
            return null;
        }
    }
} @Getter
public class SqlGremlinQueryResult {
    public static final String EMPTY_MESSAGE = "No more results.";
    public static final String NULL_VALUE = "$%#NULL#%$";
    private final List<String> columns;
    private final List<String> columnTypes = new ArrayList<>();
    private final Object assertEmptyLock = new Object();
    private final BlockingQueue<List<Object>> blockingQueueRows = new LinkedBlockingQueue<>();
    private boolean isEmpty = false;
    private SQLException paginationException = null;
    private Thread currThread = null;

    public SqlGremlinQueryResult(final List<String> columns, final List<GremlinTableBase> gremlinTableBases,
                                 final SqlMetadata sqlMetadata) throws SQLException {
        this.columns = columns;
        for (final String column : columns) {
            GremlinProperty col = null;
            for (final GremlinTableBase gremlinTableBase : gremlinTableBases) {
                if (sqlMetadata.getTableHasColumn(gremlinTableBase, column)) {
                    col = sqlMetadata.getGremlinProperty(gremlinTableBase.getLabel(), column);
                    break;
                }
            }
            columnTypes.add((col == null || col.getType() == null) ? "string" : col.getType());
        }
    }

    public void setPaginationException(final SQLException e) {
        synchronized (assertEmptyLock) {
            paginationException = e;
            if (currThread != null && blockingQueueRows.size() == 0) {
                currThread.interrupt();
            }
        }
    }

    public boolean getIsEmpty() throws SQLException {
        if (paginationException == null) {
            return isEmpty;
        }
        throw paginationException;
    }

    public void assertIsEmpty() {
        synchronized (assertEmptyLock) {
            if (currThread != null && blockingQueueRows.size() == 0) {
                currThread.interrupt();
            }
            isEmpty = true;
        }
    }

    public void addResults(final List<List<Object>> rows) {
        for (final List<Object> row : rows) {
            for (int i = 0; i < row.size(); i++) {
                if (row.get(i) instanceof String && row.get(i).toString().equals(NULL_VALUE)) {
                    row.set(i, null);
                }
            }
        }
        blockingQueueRows.addAll(rows);
    }

    private boolean getShouldExit() throws SQLException {
        synchronized (assertEmptyLock) {
            return (getIsEmpty() && blockingQueueRows.size() == 0);
        }
    }

    public Object getResult() throws SQLException {
        synchronized (assertEmptyLock) {
            this.currThread = Thread.currentThread();
        }
        while (!getShouldExit()) {
            try {
                return this.blockingQueueRows.take();
            } catch (final InterruptedException ignored) {
                if (paginationException != null) {
                    throw paginationException;
                }
            }
        }
        throw new SQLException(EMPTY_MESSAGE);
    }
} public class JoinDataReader implements GetRowFromMap {
    private final List<Pair<String, String>> tableColumnList = new ArrayList<>();

    public JoinDataReader(final Map<String, List<String>> tablesColumns) {
        tablesColumns.forEach((key, value) -> value.forEach(column -> tableColumnList.add(new Pair<>(key, column))));
    }

    @Override
    public Object[] execute(final Map<String, Object> map) {
        final Object[] row = new Object[tableColumnList.size()];
        int i = 0;
        for (final Pair<String, String> tableColumn : tableColumnList) {
            final Optional<String> tableKey =
                    map.keySet().stream().filter(key -> key.equalsIgnoreCase(tableColumn.left)).findFirst();
            if (!tableKey.isPresent()) {
                row[i++] = null;
                continue;
            }

            final Optional<String> columnKey = ((Map<String, Object>) map.get(tableKey.get())).keySet().stream()
                    .filter(key -> key.equalsIgnoreCase(tableColumn.right)).findFirst();
            if (!columnKey.isPresent()) {
                row[i++] = null;
                continue;
            }
            row[i++] = ((Map<String, Object>) map.get(tableKey.get())).getOrDefault(columnKey.get(), null);
        }
        return row;
    }
} @AllArgsConstructor
public class Pagination implements Runnable {
    private static final Logger LOGGER = LoggerFactory.getLogger(Pagination.class);
    private static final int DEFAULT_PAGE_SIZE = 1000;
    private final int pageSize = DEFAULT_PAGE_SIZE;
    private final GetRowFromMap getRowFromMap;
    private final GraphTraversal<?, ?> traversal;
    private final SqlGremlinQueryResult sqlGremlinQueryResult;

    @Override
    public void run() {
        try {
            LOGGER.info("Graph traversal: " +
                    GroovyTranslator.of("g").translate(traversal.asAdmin().getBytecode()));
            while (traversal.hasNext()) {
                final List<Object> rows = new ArrayList<>();
                traversal.next(pageSize).forEach(map -> rows.add(getRowFromMap.execute((Map<String, Object>) map)));
                convertAndInsertResult(sqlGremlinQueryResult, rows);
            }
            // If we run out of traversal data (or hit our limit), stop and signal to the result that it is done.
            sqlGremlinQueryResult.assertIsEmpty();
        } catch (final Exception e) {
            final StringWriter sw = new StringWriter();
            final PrintWriter pw = new PrintWriter(sw);
            e.printStackTrace(pw);
            LOGGER.error("Encountered exception", e);
            sqlGremlinQueryResult.setPaginationException(new SQLException(e + pw.toString()));
        }
    }

    /**
     * converts input row results and insert them into sqlGremlinQueryResult
     */
    void convertAndInsertResult(final SqlGremlinQueryResult sqlGremlinQueryResult, final List<Object> rows) {
        final List<List<Object>> finalRowResult = new ArrayList<>();
        for (final Object row : rows) {
            final List<Object> convertedRow = new ArrayList<>();
            if (row instanceof Object[]) {
                convertedRow.addAll(Arrays.asList((Object[]) row));
            } else {
                convertedRow.add(row);
            }
            finalRowResult.add(convertedRow);
        }
        sqlGremlinQueryResult.addResults(finalRowResult);
    }
}
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.calcite.sql.JoinConditionType;
import org.apache.calcite.sql.JoinType;
import org.apache.calcite.sql.SqlJoin;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.SqlSelect;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
import org.apache.tinkerpop.gremlin.structure.Column;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.twilmes.sql.gremlin.adapter.converter.SqlMetadata;
import org.twilmes.sql.gremlin.adapter.converter.SqlTraversalEngine;
import org.twilmes.sql.gremlin.adapter.converter.ast.nodes.GremlinSqlFactory;
import org.twilmes.sql.gremlin.adapter.converter.ast.nodes.GremlinSqlNode;
import org.twilmes.sql.gremlin.adapter.converter.ast.nodes.operands.GremlinSqlIdentifier;
import org.twilmes.sql.gremlin.adapter.converter.ast.nodes.operator.GremlinSqlAsOperator;
import org.twilmes.sql.gremlin.adapter.converter.ast.nodes.operator.GremlinSqlBasicCall;
import org.twilmes.sql.gremlin.adapter.converter.ast.nodes.select.join.GremlinSqlJoinComparison;
import org.twilmes.sql.gremlin.adapter.converter.schema.gremlin.GremlinTableBase;
import org.twilmes.sql.gremlin.adapter.results.SqlGremlinQueryResult;
import org.twilmes.sql.gremlin.adapter.results.pagination.JoinDataReader;
import org.twilmes.sql.gremlin.adapter.results.pagination.Pagination;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/*
 * 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.twilmes.sql.gremlin.adapter.converter.ast.nodes.select;


/**
 * This module is a GremlinSql equivalent of Calcite's SqlSelect for a JOIN operation.
 *
 * @author Lyndon Bauto (lyndonb@bitquilltech.com)
 */
public class GremlinSqlSelectMulti extends GremlinSqlSelect {
    // Multi is a JOIN.
    private static final Logger LOGGER = LoggerFactory.getLogger(GremlinSqlSelectMulti.class);
    private final SqlSelect sqlSelect;
    private final SqlMetadata sqlMetadata;
    private final GraphTraversalSource g;
    private final SqlJoin sqlJoin;

    public GremlinSqlSelectMulti(final SqlSelect sqlSelect, final SqlJoin sqlJoin,
                                 final SqlMetadata sqlMetadata, final GraphTraversalSource g) {
        super(sqlSelect, sqlMetadata);
        this.sqlMetadata = sqlMetadata;
        this.sqlSelect = sqlSelect;
        this.g = g;
        this.sqlJoin = sqlJoin;
    }

    @Override
    protected void runTraversalExecutor(final GraphTraversal<?, ?> graphTraversal,
                                        final SqlGremlinQueryResult sqlGremlinQueryResult) throws SQLException {
        // Launch thread to continue grabbing results.
        final ExecutorService executor = Executors.newSingleThreadExecutor(
                new ThreadFactoryBuilder().setNameFormat("Data-Insert-Thread-%d").setDaemon(true).build());
        final Map<String, List<String>> tableColumns = sqlMetadata.getColumnOutputListMap();
        if (tableColumns.keySet().size() > 2) {
            throw new SQLException("Error: Join expects one or two tables only.");
        }
        executor.execute(new Pagination(new JoinDataReader(tableColumns), graphTraversal, sqlGremlinQueryResult));
        executor.shutdown();
    }

    @Override
    public GraphTraversal<?, ?> generateTraversal() throws SQLException {
        final JoinType joinType = sqlJoin.getJoinType();
        final JoinConditionType conditionType = sqlJoin.getConditionType();

        final GremlinSqlBasicCall left =
                GremlinSqlFactory.createNodeCheckType(sqlJoin.getLeft(), GremlinSqlBasicCall.class);