odps/df/backends/odpssql/compiler.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright 1999-2022 Alibaba Group Holding Ltd.
#
# 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.

import re
from datetime import date, datetime
from decimal import Decimal

from ...expr.reduction import *
from ...expr.arithmetic import BinOp, Add, Substract, Power, Invert, Negate, Abs
from ...expr.merge import JoinCollectionExpr, UnionCollectionExpr
from ...expr.element import MappedExpr, Func
from ...expr.window import CumSum, NthValue, QCut
from ...expr.datetimes import DTScalar
from ...expr.collections import RowAppliedCollectionExpr
from ...expr import element, strings, datetimes, composites
from ...utils import is_source_collection, traverse_until_source
from ... import types as df_types
from . import types
from .models import MemCacheReference
from ..core import Backend
from .... import utils
from ....models import Function
from ..errors import CompileError


BINARY_OP_COMPILE_DIC = {
    'Add': '+',
    'Substract': '-',
    'Multiply': '*',
    'Divide': '/',
    'Greater': '>',
    'GreaterEqual': '>=',
    'Less': '<',
    'LessEqual': '<=',
    'Equal': '==',
    'NotEqual': '!=',
    'And': 'and',
    'Or': 'or'
}

UNARY_OP_COMPILE_DIC = {
    'Negate': '-'
}

WINDOW_COMPILE_DIC = {
    'CumSum': 'sum',
    'CumMean': 'avg',
    'CumMedian': 'median',
    'CumStd': 'stddev',
    'CumMax': 'max',
    'CumMin': 'min',
    'CumCount': 'count',
    'NthValue': 'nth_value',
    'Lag': 'lag',
    'Lead': 'lead',
    'Rank': 'rank',
    'DenseRank': 'dense_rank',
    'PercentRank': 'percent_rank',
    'RowNumber': 'row_number',
    'QCut': 'ntile',
    'CumeDist': 'cume_dist',
}

MATH_COMPILE_DIC = {
    'Abs': 'abs',
    'Sqrt': 'sqrt',
    'Sin': 'sin',
    'Sinh': 'sinh',
    'Cos': 'cos',
    'Cosh': 'cosh',
    'Tan': 'tan',
    'Tanh': 'tanh',
    'Exp': 'exp',
    'Arccos': 'acos',
    'Arcsin': 'asin',
    'Arctan': 'atan',
    'Ceil': 'ceil',
    'Floor': 'floor'
}

DATE_PARTS_DIC = {
    'Year': 'yyyy',
    'Month': 'mm',
    'Day': 'dd',
    'Hour': 'hh',
    'Minute': 'mi',
    'Second': 'ss'
}


class OdpsSQLCompiler(Backend):
    """
    OdpsSQLCompiler will compile an Expr into an ODPS SQL.
    """

    def __init__(self, ctx, indent_size=2, beautify=False):
        self._ctx = ctx
        self._indent_size = indent_size
        self._beautify = beautify

        # use for `join` or `union` operations etc.
        self._sub_compiles = defaultdict(lambda: list())
        self._union_no_alias = dict()
        # When encountering `join` or `union`, we will try to compile all child branches,
        # for each nodes of these branches, we should not check the uniqueness,
        # when compilation finishes, we substitute the children of `join` or `union` with None,
        # so the upcoming compilation will not visit its children.
        # When everything are done, we use the callbacks to substitute back the original children
        # of the `join` or `union` node.
        self._callbacks = list()

        # store the expr ids to do mem cache
        self._mem_ref_caches = set()

        self._re_init()

    def _re_init(self):
        self._select_clause = None
        self._from_clause = None
        self._where_clause = None
        self._group_by_clause = None
        self._having_clause = None
        self._order_by_clause = None
        self._table_sample_clause = None
        self._limit = None

    def _cleanup(self):
        self._sub_compiles = defaultdict(lambda: list())
        for callback in self._callbacks:
            callback()
        self._callbacks = list()

    @classmethod
    def _need_recursive_handle_in_expr(cls, node):
        return isinstance(node, (element.IsIn, element.NotIn)) and \
               not all(n is None for n in node.args) and \
               isinstance(node.values[0], SequenceExpr)

    @classmethod
    def _retrieve_until_find_root(cls, expr):
        for node in traverse_until_source(expr, top_down=True, unique=True):
            if isinstance(node, (JoinCollectionExpr, UnionCollectionExpr, LateralViewCollectionExpr)) and \
                    not all(n is None for n in node.args):
                yield node
            else:
                ins = [n for n in node.children() if cls._need_recursive_handle_in_expr(n)]
                if len(ins) > 0:
                    for n in ins:
                        yield n

    def _compile_union_node(self, expr, traversed):
        if isinstance(expr.lhs, UnionCollectionExpr):
            self._union_no_alias[expr.lhs] = True
        compiled = self._compile(expr.lhs)

        self._sub_compiles[expr].append(compiled)

        compiled = self._compile(expr.rhs)
        self._sub_compiles[expr].append(compiled)

        args = expr.args
        self._ctx._expr_raw_args[id(expr)] = args
        for arg_name in expr._args:
            setattr(expr, arg_name, None)

        def cb():
            for arg_name, arg in zip(expr._args, args):
                setattr(expr, arg_name, arg)

        self._callbacks.append(cb)

    def _compile_join_node(self, expr, traversed):
        travs = set()

        compiled, trav = self._compile(expr.lhs, return_traversed=True)
        travs.update(trav)
        if not is_source_collection(expr.lhs) and not isinstance(expr.lhs, JoinCollectionExpr):
            self._sub_compiles[expr].append(
                '(\n{0}\n) {1}'.format(utils.indent(compiled, self._indent_size),
                                     self._ctx.get_collection_alias(expr.lhs, create=True)[0])
            )
        else:
            self._sub_compiles[expr].append(self._ctx.get_expr_compiled(expr.lhs))

        compiled, trav = self._compile(expr.rhs, return_traversed=True)
        travs.update(trav)
        if not is_source_collection(expr.rhs):
            self._sub_compiles[expr].append(
                '(\n{0}\n) {1}'.format(utils.indent(compiled, self._indent_size),
                                     self._ctx.get_collection_alias(expr.rhs, create=True)[0])
            )
        else:
            self._sub_compiles[expr].append(self._ctx.get_expr_compiled(expr.rhs))

        if expr.predicate is None:
            self._sub_compiles[expr].append(None)
            traversed.update(travs)
        else:
            self._compile(expr.predicate, traversed)
            self._sub_compiles[expr].append(self._ctx.get_expr_compiled(expr.predicate))

        if expr._mapjoin:
            self._ctx._mapjoin_hints.append(self._ctx.get_collection_alias(expr.rhs)[0])

        if expr._skewjoin:
            skewjoin_expr = self._ctx.get_collection_alias(expr.rhs)[0]
            if isinstance(expr._skewjoin, list):
                skewjoin_expr += '({0})'.format(', '.join(expr._skewjoin))
            if expr._skewjoin_values:
                skewjoin_expr += '({0})'.format(
                    ', '.join(
                        '({0})'.format(', '.join(repr(s) for s in values))
                        for values in expr._skewjoin_values
                    )
                )
            self._ctx._skewjoin_hints.append(skewjoin_expr)

        args = expr.args
        self._ctx._expr_raw_args[id(expr)] = args
        for arg_name in expr._args:
            setattr(expr, arg_name, None)

        def cb():
            for arg_name, arg in zip(expr._args, args):
                setattr(expr, arg_name, arg)

        self._callbacks.append(cb)

    @classmethod
    def _find_table(cls, expr):
        return next(it for it in expr.traverse(top_down=True, unique=True)
                    if isinstance(it, CollectionExpr))

    def _compile_in_node(self, expr, traversed):
        self._compile(expr.input)
        self._sub_compiles[expr].append(self._ctx.get_expr_compiled(expr.input))

        to_sub = self._find_table(expr.values[0])[[expr.values[0], ]]
        compiled = self._compile(to_sub)
        self._sub_compiles[expr].append(compiled)

        args = expr.args
        self._ctx._expr_raw_args[id(expr)] = args
        for arg_name in expr._args:
            setattr(expr, arg_name, None)

        def cb():
            for arg_name, arg in zip(expr._args, args):
                setattr(expr, arg_name, arg)

        self._callbacks.append(cb)

    def _compile_lateral_view(self, expr, traversed):
        travs = set()
        compiled_input, trav = self._compile(expr.input, return_traversed=True)
        travs.update(trav)

        if not is_source_collection(expr.input) and not isinstance(expr.input, LateralViewCollectionExpr):
            compiled_input = '(\n{0}\n) {1}'.format(utils.indent(compiled_input, self._indent_size),
                                                    self._ctx.get_collection_alias(expr.input, create=True)[0])
        else:
            compiled_input = self._ctx.get_expr_compiled(expr.input)

        from_lines = [compiled_input]
        for lview in expr.lateral_views:
            self._compile(lview)
            from_lines.append(self._ctx.get_expr_compiled(lview))
        from_clause = ' \n'.join(from_lines)

        traversed.update(travs)

        self._sub_compiles[expr].append(from_clause)

        args = expr.args
        self._ctx._expr_raw_args[id(expr)] = args
        for arg_name in expr._args:
            setattr(expr, arg_name, None)

        def cb():
            for arg_name, arg in zip(expr._args, args):
                setattr(expr, arg_name, arg)

        self._callbacks.append(cb)

    def _compile(self, expr, traversed=None,
                 return_traversed=False, root_expr=None):
        roots = self._retrieve_until_find_root(expr)

        if traversed is None:
            traversed = set()

        for root in roots:
            if root is not None:
                if isinstance(root, JoinCollectionExpr):
                    self._compile_join_node(root, traversed)
                elif isinstance(root, UnionCollectionExpr):
                    self._compile_union_node(root, traversed)
                elif isinstance(root, LateralViewCollectionExpr):
                    self._compile_lateral_view(root, traversed)
                elif isinstance(root, (element.IsIn, element.NotIn)):
                    self._compile_in_node(root, traversed)
                root.accept(self)
                traversed.add(id(root))

        for node in traverse_until_source(expr):
            if id(node) not in traversed:
                node.accept(self)
                traversed.add(id(node))

        if (
            expr is root_expr
            and self._select_clause is None
            and (self._ctx._mapjoin_hints or self._ctx._skewjoin_hints)
        ):
            self.add_select_clause(expr, '* ')

        sql = self.to_sql().strip()
        if not return_traversed:
            return sql
        return sql, traversed

    def compile(self, expr):
        try:
            sql = self._compile(expr, root_expr=expr)
            symbol_columns = dict(self._ctx.get_all_need_alias_column_symbols())
            sql = self._fill_back_columns(sql, symbol_columns)
            sql = self._re_join_select_fields(sql, symbol_columns)

            if self._mem_ref_caches:
                dep_sqls = self._ctx.get_mem_cache_dep_sqls(*self._mem_ref_caches)
                dep_sqls = [dep_sql if dep_sql.endswith(';') else dep_sql + ';'
                            for dep_sql in dep_sqls]
                return dep_sqls + [sql,]

            return sql
        finally:
            self._cleanup()

    def to_sql(self):
        if not self._from_clause.startswith('SELECT'):
            lines = [
                'SELECT {0} '.format(self._select_clause or '*'),
                'FROM {0} '.format(self._from_clause),
            ]
        else:
            # special case due to `union`
            lines = [self._from_clause]

        # tablesample need to be placed right after from clause
        if self._table_sample_clause:
            lines.append('TABLESAMPLE {0}'.format(self._table_sample_clause))
        if self._where_clause:
            lines.append('WHERE {0} '.format(self._where_clause))
        if self._group_by_clause:
            lines.append(self._group_by_clause)
            if self._having_clause:
                lines.append('HAVING {0} '.format(self._having_clause))
        if self._order_by_clause:
            if self._order_by_clause.startswith('ORDER BY') and not self._limit and \
                    options.df.odps.sort.limit:
                # for `order by`, limit is required
                # for `sort by`, limit is unnecessary
                self._limit = options.df.odps.sort.limit
            lines.append(self._order_by_clause)
        if self._limit is not None:
            lines.append('LIMIT {0}'.format(self._limit))

        self._re_init()
        return '\n'.join(lines)

    def _fill_back_columns(self, sql, symbols_to_columns):
        symbol_compiled = dict()
        for symbol, column in six.iteritems(symbols_to_columns):
            try:
                collection, name = self._retrieve_column_alias_collection(column)
            except KeyError:
                # sorted column relative, just ignore
                continue
            symbol_compiled[symbol] = '{0}.{1}'.format(
                self._ctx.get_collection_alias(collection)[0],
                self._quote(name))
        sql = sql % symbol_compiled

        reg = re.compile(r'###\[(col_\d+)\]##')

        def repl(matched):
            symbol = matched.group(1)
            compiled = symbol_compiled[symbol]
            compiled_name = self._unquote(compiled)
            column = symbols_to_columns[symbol]
            if compiled_name != column.name:
                return ' AS {0}'.format(self._quote(column.name))
            else:
                return ''

        return reg.sub(repl, sql)

    def _re_join_select_fields(self, sql, symbols_to_columns):
        if not self._beautify:
            return sql

        reg = re.compile(r'/{_i(\d+)}')

        for select_idx in reg.findall(sql):
            s_regex_str = '//{{_i{0}}}(.+?){{_i{0}}}//'.format(select_idx)
            regex_str = '\n?( *)/{{_i{0}}}({1})+/'.format(select_idx, s_regex_str)
            regex = re.compile(regex_str, (re.M | re.DOTALL))
            s_regex = re.compile(s_regex_str, (re.M | re.DOTALL))

            def repl(matched):
                space = len(matched.group(1))
                joined = matched.group()

                fields = s_regex.findall(joined)
                sub = self._join_compiled_fields(fields)
                if not joined.startswith('\n'):
                    sub = sub.lstrip('\n')
                if space > self._indent_size:
                    return utils.indent(sub, space-self._indent_size)
                return sub

            sql = regex.sub(repl, sql)

        return sql

    def _retrieve_column_alias_collection(self, expr):
        column_name = expr.source_name

        collection = expr.input
        while True:
            if isinstance(collection, JoinCollectionExpr):
                idx, column_name = collection._column_origins[column_name]
                args = self._ctx._expr_raw_args[id(collection)]  # get the args which are substituted out
                lhs, rhs = args[0], args[1]
                collection = (lhs, rhs)[idx]
            elif isinstance(collection, LateralViewCollectionExpr):
                args = self._ctx._expr_raw_args[id(collection)]  # get the args which are substituted out
                while True:
                    for lv in args[2]:
                        if utils.to_lower_str(column_name) in lv.schema._name_indexes:
                            return lv, column_name
                    if isinstance(args[0], LateralViewCollectionExpr):
                        args = self._ctx._expr_raw_args[id(args[0])]
                    else:
                        return args[0], column_name
            elif self._ctx.get_collection_alias(collection, silent=True):
                return collection, column_name
            elif isinstance(getattr(collection, 'input', None), CollectionExpr):
                collection = collection.input

        raise CompileError('Cannot find table alias for column: \n%s' % repr_obj(expr))

    def sub_sql_to_from_clause(self, expr):
        sql = self.to_sql()

        alias, _ = self._ctx.get_collection_alias(expr, create=True)
        from_clause = '(\n{0}\n) {1}'.format(
            utils.indent(sql, self._indent_size), alias
        )

        self._re_init()
        self._from_clause = from_clause

    def add_select_clause(self, expr, select_clause):
        if self._select_clause is not None:
            self.sub_sql_to_from_clause(expr.input)
        elif self._order_by_clause is not None:
            self.sub_sql_to_from_clause(expr.input)
        elif isinstance(expr, Summary) and self._limit is not None:
            self.sub_sql_to_from_clause(expr.input)
        elif isinstance(expr, (GroupByCollectionExpr, MutateCollectionExpr, RowAppliedCollectionExpr)) and \
                self._limit is not None:
            self.sub_sql_to_from_clause(expr.input)
        elif isinstance(expr.input, ReshuffledCollectionExpr):
            self.sub_sql_to_from_clause(expr.input)

        self._select_clause = select_clause
        join_hints = []
        if self._ctx._mapjoin_hints:
            join_hints.append(
                'mapjoin({0})'.format(', '.join(self._ctx._mapjoin_hints))
            )
            self._ctx._mapjoin_hints = []
        if self._ctx._skewjoin_hints:
            join_hints.append(
                'skewjoin({0})'.format(', '.join(self._ctx._skewjoin_hints))
            )
            self._ctx._skewjoin_hints = []

        if join_hints:
            self._select_clause = '/*+ {0} */ {1}'.format(
                ', '.join(join_hints), self._select_clause
            )

    def add_from_clause(self, expr, from_clause):
        if self._from_clause is None:
            self._from_clause = from_clause

    def add_where_clause(self, expr, where_clause):
        if any(clause is not None for clause in
               (self._where_clause, self._select_clause, self._limit)):
            self.sub_sql_to_from_clause(expr.input)

        self._where_clause = where_clause

    def add_group_by_clause(self, expr, group_by_clause):
        if self._group_by_clause is not None:
            self.sub_sql_to_from_clause(expr.input)
        elif isinstance(expr, ReshuffledCollectionExpr) and \
                self._select_clause is not None:
            self.sub_sql_to_from_clause(expr.input)

        self._group_by_clause = group_by_clause

    def add_having_clause(self, expr, having_clause):
        if self._having_clause is None:
            self._having_clause = having_clause

        assert having_clause == self._having_clause

    def add_order_by_clause(self, expr, order_by_clause):
        if self._order_by_clause is not None:
            self.sub_sql_to_from_clause(expr.input)

        self._order_by_clause = order_by_clause

    def add_table_sample_clause(self, expr, table_sample_clause):
        if any(clause is not None for clause in
               (self._where_clause, self._table_sample_clause, self._group_by_clause,
                self._having_clause, self._order_by_clause, self._limit)):
            self.sub_sql_to_from_clause(expr.input)
        self._table_sample_clause = table_sample_clause

    def set_limit(self, expr, limit):
        if self._limit is not None:
            self.sub_sql_to_from_clause(expr.input)

        self._limit = limit

    def visit_source_collection(self, expr):
        source_data = expr._source_data
        alias = self._ctx.register_collection(expr)
        if isinstance(source_data, MemCacheReference):
            from_clause = '{0} {1}'.format(source_data.ref_name, alias)
            self.add_from_clause(expr, from_clause)
            self._ctx.add_expr_compiled(expr, from_clause)
            self._mem_ref_caches.add(source_data.expr_id)
        else:
            table_parts = [source_data.project.name]
            schema = source_data.get_schema()
            if schema is not None:
                table_parts.append(schema.name)

            if options.df.quote:
                table_parts.append("`%s`" % source_data.name)
            else:
                table_parts.append(source_data.name)

            name = '.'.join(table_parts)

            from_clause = '{0} {1}'.format(name, alias)
            self.add_from_clause(expr, from_clause)
            self._ctx.add_expr_compiled(expr, from_clause)

    def _compile_select_field(self, field):
        compiled = self._ctx.get_expr_compiled(field)

        if not isinstance(field, Column) or field._source_data_type != field._data_type:
            compiled = '{0} AS {1}'.format(compiled, self._quote(field.name))
        else:
            if compiled.startswith('%(') and compiled.endswith(')s'):
                symbol = compiled[2:-2]
                compiled = '{0}###[{1}]##'.format(compiled, symbol)
            else:
                compiled_name = self._unquote(compiled)
                if field.name != compiled_name:
                    compiled = '{0} AS {1}'.format(compiled, self._quote(field.name))

        return compiled

    def _join_select_fields(self, fields):
        if not self._beautify:
            return ', '.join(fields)
        else:
            select_id = self._ctx.next_select_id()

            def h(field):
                return '//{{_i{0}}}{1}{{_i{0}}}//'.format(select_id, field)

            return utils.indent('\n/{{_i{0}}}{1}/'.format(select_id, ''.join([h(f) for f in fields])),
                                self._indent_size)

    def _join_compiled_fields(self, fields):
        if not self._beautify:
            return ', '.join(fields)
        else:
            buf = six.StringIO()
            buf.write('\n')

            split_fields = [field.rsplit(' AS ', 1) for field in fields]
            get = lambda s: s if '\n' not in s else s.rsplit('\n', 1)[1]
            max_length = max(len(get(f[0])) for f in split_fields)

            for f in split_fields:
                if len(f) > 1:
                    buf.write(f[0].ljust(max_length))
                    buf.write(' AS ')
                    buf.write(f[1])
                else:
                    buf.write(f[0])
                buf.write(',\n')

            return utils.indent(buf.getvalue()[:-2], self._indent_size)

    def visit_project_collection(self, expr):
        fields = expr._fields
        compiled_fields = [self._compile_select_field(field)
                           for field in fields]

        compiled = self._join_select_fields(compiled_fields)

        self._ctx.add_expr_compiled(expr, compiled)

        self.add_select_clause(expr, compiled)

    def visit_lateral_view(self, expr):
        compiled = self._sub_compiles[expr][0]
        self._ctx.add_expr_compiled(expr, compiled)
        self.add_from_clause(expr, compiled)

    def visit_filter_collection(self, expr):
        predicate = expr.args[1]
        compiled = self._ctx.get_expr_compiled(predicate)

        self._ctx.add_expr_compiled(expr, compiled)
        self.add_where_clause(expr, compiled)

    def visit_filter_partition_collection(self, expr):
        compiled = self._ctx.get_expr_compiled(expr._predicate)

        self._ctx.add_expr_compiled(expr, compiled)
        self.add_where_clause(expr, compiled)

        compiled_fields = [self._compile_select_field(field)
                           for field in expr.fields]

        compiled = self._join_select_fields(compiled_fields)

        self._ctx.add_expr_compiled(expr, compiled)
        self.add_select_clause(expr, compiled)

    def visit_slice_collection(self, expr):
        sliced = expr._indexes
        if sliced[0] is not None:
            raise NotImplementedError
        if sliced[2] is not None:
            raise NotImplementedError
        if sliced[1].value <= 0:
            raise CompileError('limit number must be greater than 0')

        self.set_limit(expr, sliced[1].value)

    def visit_element_op(self, expr):
        if isinstance(expr, element.IsNull):
            compiled = '{0} IS NULL'.format(self._parenthesis(expr.input))
        elif isinstance(expr, element.NotNull):
            compiled = '{0} IS NOT NULL'.format(self._parenthesis(expr.input))
        elif isinstance(expr, element.FillNa):
            compiled = 'IF(%(input)s IS NULL, %(value)s, %(else_value)s)' % {
                'input': self._parenthesis(expr.input),
                'value': self._ctx.get_expr_compiled(expr._fill_value),
                'else_value': self._ctx.get_expr_compiled(expr.input),
            }
        elif isinstance(expr, element.IsIn):
            if expr.values is not None:
                compiled = '{0} IN ({1})'.format(
                    self._ctx.get_expr_compiled(expr.input),
                    ', '.join(self._ctx.get_expr_compiled(it) for it in expr.values)
                )
            else:
                subs = self._sub_compiles[expr]
                compiled = '{0} IN ({1})'.format(
                    subs[0], subs[1].replace('\n', '')
                )
        elif isinstance(expr, element.NotIn):
            if expr.values is not None:
                compiled = '{0} NOT IN ({1})'.format(
                    self._ctx.get_expr_compiled(expr.input),
                    ', '.join(self._ctx.get_expr_compiled(it) for it in expr.values)
                )
            else:
                subs = self._sub_compiles[expr]
                compiled = '{0} NOT IN ({1})'.format(
                    subs[0], subs[1].replace('\n', '')
                )
        elif isinstance(expr, element.IfElse):
            compiled = 'IF({0}, {1}, {2})'.format(
                self._ctx.get_expr_compiled(expr._input),
                self._ctx.get_expr_compiled(expr._then),
                self._ctx.get_expr_compiled(expr._else),
            )
        elif isinstance(expr, element.Switch):
            case = self._ctx.get_expr_compiled(expr.case) + ' ' \
                if expr.case is not None else ''
            lines = ['CASE {0}'.format(case)]
            for pair in zip(expr.conditions, expr.thens):
                args = [self._ctx.get_expr_compiled(p) for p in pair]
                lines.append('WHEN {0} THEN {1} '.format(*args))
            if expr.default is not None:
                lines.append('ELSE {0} '.format(self._ctx.get_expr_compiled(expr.default)))
            lines.append('END')
            if self._beautify:
                for i in range(1, len(lines) - 1):
                    lines[i] = utils.indent(lines[i], self._indent_size)
                compiled = '\n'.join(lines)
            else:
                compiled = ''.join(lines)
        elif isinstance(expr, element.IntToDatetime):
            compiled = 'FROM_UNIXTIME({0})'.format(
                self._ctx.get_expr_compiled(expr._input),
            )
        else:
            raise NotImplementedError

        self._ctx.add_expr_compiled(expr, compiled)

    def _parenthesis(self, child):
        if isinstance(child, BinOp):
            return '(%s)' % self._ctx.get_expr_compiled(child)
        elif isinstance(child, (element.IsNull, element.NotNull, element.IsIn, element.NotIn,
                                element.Between, element.Switch, element.Cut)):
            return '(%s)' % self._ctx.get_expr_compiled(child)
        else:
            return self._ctx.get_expr_compiled(child)

    def visit_binary_op(self, expr):
        if isinstance(expr, Add) and expr.dtype == df_types.string:
            compiled = 'CONCAT({0}, {1})'.format(
                self._ctx.get_expr_compiled(expr.lhs),
                self._ctx.get_expr_compiled(expr.rhs),
            )
        elif isinstance(expr, (Add, Substract)) and expr.dtype == df_types.datetime:
            if isinstance(expr, Add) and \
                    all(child.dtype == df_types.datetime for child in (expr.lhs, expr.rhs)):
                raise CompileError('Cannot add two datetimes')
            if isinstance(expr.rhs, DTScalar) or (isinstance(expr, Add) and expr.lhs, DTScalar):
                if isinstance(expr.rhs, DTScalar):
                    dt, scalar = expr.lhs, expr.rhs
                else:
                    dt, scalar = expr.rhs, expr.lhs

                class_name = type(scalar).__name__[:-6]
                date_part = DATE_PARTS_DIC[class_name]
                val = scalar.value
                if isinstance(expr, Substract):
                    val = -val
                compiled = 'DATEADD({0}, {1}, {2})'.format(
                    self._ctx.get_expr_compiled(dt), repr(val), repr(date_part)
                )
        else:
            compiled, op = None, None
            try:
                op = BINARY_OP_COMPILE_DIC[expr.node_name].upper()
            except KeyError:
                if isinstance(expr, Power):
                    compiled = 'POW({0}, {1})'.format(
                        self._ctx.get_expr_compiled(expr.lhs),
                        self._ctx.get_expr_compiled(expr.rhs)
                    )
                    if not isinstance(expr.dtype, df_types.Float):
                        compiled = self._cast(compiled, df_types.float64, expr.dtype)
                else:
                    raise NotImplementedError

            if compiled is None:
                lhs, rhs = expr.args
                if op:
                    compiled = '{0} {1} {2}'.format(
                        self._parenthesis(lhs), op, self._parenthesis(rhs)
                    )
                else:
                    raise NotImplementedError

        self._ctx.add_expr_compiled(expr, compiled)

    def visit_unary_op(self, expr):
        try:
            if isinstance(expr, Negate) and expr.input.dtype == df_types.boolean:
                compiled = 'NOT {0}'.format(self._parenthesis(expr.input))
            else:
                op = UNARY_OP_COMPILE_DIC[expr.node_name]

                compiled = '{0}{1}'.format(
                        op, self._parenthesis(expr.input))
        except KeyError:
            if isinstance(expr, Abs):
                compiled = 'ABS({0})'.format(
                    self._ctx.get_expr_compiled(expr.input))
            elif isinstance(expr, (Invert, Negate)) and \
                    expr.input.dtype == df_types.boolean:
                compiled = 'NOT {0}'.format(self._parenthesis(expr.input))
            else:
                raise NotImplementedError

        self._ctx.add_expr_compiled(expr, compiled)

    def visit_math(self, expr):
        compiled = None
        try:
            op = MATH_COMPILE_DIC[expr.node_name]
        except KeyError:
            if expr.node_name == 'Log':
                if expr._base is None:
                    op = 'ln'
                else:
                    compiled = 'LOG({0}, {1})'.format(
                        self._ctx.get_expr_compiled(expr._base),
                        self._ctx.get_expr_compiled(expr.input)
                    )
            elif expr.node_name == 'Log2':
                compiled = 'LOG(2, {0})'.format(
                    self._ctx.get_expr_compiled(expr.input)
                )
            elif expr.node_name == 'Log10':
                compiled = 'LOG(10, {0})'.format(
                    self._ctx.get_expr_compiled(expr.input)
                )
            elif expr.node_name == 'Log1p':
                compiled = 'LN(1 + {0})'.format(
                    self._ctx.get_expr_compiled(expr.input)
                )
            elif expr.node_name == 'Expm1':
                compiled = 'EXP({0}) - 1'.format(
                    self._ctx.get_expr_compiled(expr.input)
                )
            elif expr.node_name == 'Trunc':
                if expr._decimals is None:
                    op = 'TRUNC'
                else:
                    compiled = 'TRUNC({0}, {1})'.format(
                        self._ctx.get_expr_compiled(expr.input),
                        self._ctx.get_expr_compiled(expr._decimals)
                    )
            elif expr.node_name == 'Round':
                if expr._decimals is None:
                    op = 'ROUND'
                else:
                    compiled = 'ROUND({0}, {1})'.format(
                        self._ctx.get_expr_compiled(expr.input),
                        self._ctx.get_expr_compiled(expr._decimals)
                    )
            else:
                raise NotImplementedError

        if compiled is None:
            compiled = '{0}({1})'.format(
                op.upper(), self._ctx.get_expr_compiled(expr.input))

        self._ctx.add_expr_compiled(expr, compiled)

    def visit_string_op(self, expr):
        # FIXME quite a few operations cannot support by internal function
        compiled = None

        input = self._ctx.get_expr_compiled(expr.input)
        if isinstance(expr, strings.Capitalize):
            compiled = 'CONCAT(TOUPPER(SUBSTR(%(input)s, 1, 1)), TOLOWER(SUBSTR(%(input)s, 2)))' % {
                'input': input
            }
        elif isinstance(expr, strings.CatStr):
            nodes = [expr._input]
            if expr._others is not None:
                others = (expr._others, ) if not isinstance(expr._others, Iterable) else expr._others
                for other in others:
                    if expr._sep is not None:
                        nodes.extend([expr._sep, other])
                    else:
                        nodes.append(other)
            compiled = 'CONCAT(%s)' % ', '.join(self._ctx.get_expr_compiled(e) for e in nodes)
        elif isinstance(expr, strings.Contains):
            if expr.regex:
                raise NotImplementedError
            compiled = 'INSTR(%s, %s) > 0' % (input, self._ctx.get_expr_compiled(expr._pat))
        elif isinstance(expr, strings.Endswith):
            # TODO: any better solution?
            compiled = 'INSTR(REVERSE(%s), REVERSE(%s)) == 1' % (
                input, self._ctx.get_expr_compiled(expr._pat))
        elif isinstance(expr, strings.Startswith):
            compiled = 'INSTR(%s, %s) == 1' % (input, self._ctx.get_expr_compiled(expr._pat))
        elif isinstance(expr, strings.Find):
            if isinstance(expr.start, six.integer_types):
                start = expr.start + 1 if expr.start >= 0 else expr.start
            else:
                start = 'IF(%(start)s >= 0, %(start)s + 1, %(start)s)' % {
                    'start': self._ctx.get_expr_compiled(expr._start)
                }
            if expr.end is not None:
                raise NotImplementedError
            else:
                compiled = 'INSTR(%s, %s, %s) - 1' % (
                    input, self._ctx.get_expr_compiled(expr._sub), start)
        elif isinstance(expr, strings.Get):
            compiled = 'SUBSTR(%s, %s, 1)' % (input, expr.index + 1)
        elif isinstance(expr, strings.Len):
            compiled = 'LENGTH(%s)' % input
        elif isinstance(expr, strings.Lower):
            compiled = 'TOLOWER(%s)' % input
        elif isinstance(expr, strings.Upper):
            compiled = 'TOUPPER(%s)' % input
        elif isinstance(expr, (strings.Lstrip, strings.Rstrip, strings.Strip)):
            if expr.to_strip != ' ':
                raise NotImplementedError
            func = {
                'Lstrip': 'LTRIM',
                'Rstrip': 'RTRIM',
                'Strip': 'TRIM'
            }
            compiled = '%s(%s)' % (func[type(expr).__name__], input)
        elif isinstance(expr, strings.Slice):
            # internal function will be compiled in two cases:
            # 1) start is not None
            # 2) positive start and end
            if expr.end is None and expr.step is None:
                compiled = 'SUBSTR(%s, %s)' % (input, expr.start + 1)
            else:
                # expr.start and expr.end
                length = expr.end - expr.start
                compiled = 'SUBSTR(%s, %s, %s)' % (input, expr.start + 1, length)
        elif isinstance(expr, strings.Repeat):
            compiled = 'REPEAT(%s, %s)' % (
                input, self._ctx.get_expr_compiled(expr._repeats))
        elif isinstance(expr, strings.Split):
            if expr.n != -1:
                raise NotImplementedError

            escape_pat = re.escape(expr.pat)
            nre_compiled = 'SPLIT(%s, \'%s\')' % (input, utils.escape_odps_string(expr.pat))
            re_compiled = 'SPLIT(%s, \'%s\')' % (input, utils.escape_odps_string(re.escape(expr.pat)))

            try:
                re.compile(expr.pat)
                is_regex = True
            except:
                is_regex = False

            if expr.pat == escape_pat or not is_regex:
                compiled = nre_compiled
            else:
                compiled = 'IF(SIZE(%(re)s) = 0, %(nre)s, %(re)s)' % dict(re=re_compiled, nre=nre_compiled)
        elif isinstance(expr, strings.StringToDict):
            compiled = 'STR_TO_MAP(%s, \'%s\', \'%s\')' % (input, expr.item_delim, expr.kv_delim)

        if compiled is not None:
            self._ctx.add_expr_compiled(expr, compiled)
        else:
            raise NotImplementedError

    def visit_datetime_op(self, expr):
        # FIXME quite a few operations cannot support by internal function
        class_name = type(expr).__name__
        input = self._ctx.get_expr_compiled(expr.input)

        compiled = None
        if class_name in DATE_PARTS_DIC:
            compiled = 'DATEPART(%s, %r)' % (input, DATE_PARTS_DIC[class_name])
        elif isinstance(expr, datetimes.WeekOfYear):
            compiled = 'WEEKOFYEAR(%s)' % input
        elif isinstance(expr, datetimes.WeekDay):
            compiled = 'WEEKDAY(%s)' % input
        elif isinstance(expr, datetimes.Date):
            compiled = 'DATETRUNC(%s, %r)' % (input, 'dd')
        elif isinstance(expr, datetimes.UnixTimestamp):
            compiled = 'UNIX_TIMESTAMP(%s)' % input

        if compiled is not None:
            self._ctx.add_expr_compiled(expr, compiled)
        else:
            raise NotImplementedError

    def visit_composite_op(self, expr):
        if getattr(expr, 'input', None) is not None:
            input = self._ctx.get_expr_compiled(expr.input)
        else:
            input = None

        compiled = None
        if isinstance(expr, composites.ListDictLength):
            compiled = 'SIZE(%s)' % input
        elif isinstance(expr, composites.ListDictGetItem):
            compiled = '%s[%s]' % (input, self._ctx.get_expr_compiled(expr._key))
        elif isinstance(expr, composites.ListSort):
            compiled = 'SORT_ARRAY(%s)' % input
        elif isinstance(expr, composites.ListContains):
            compiled = 'ARRAY_CONTAINS(%s, %s)' % (
                input,
                self._ctx.get_expr_compiled(expr._value)
            )
        elif isinstance(expr, composites.DictKeys):
            compiled = 'MAP_KEYS(%s)' % input
        elif isinstance(expr, composites.DictValues):
            compiled = 'MAP_VALUES(%s)' % input
        elif isinstance(expr, composites.ListBuilder):
            compiled = 'ARRAY(%s)' % ', '.join(
                self._ctx.get_expr_compiled(it) for it in expr._values
            )
        elif isinstance(expr, composites.DictBuilder):
            compiled = 'MAP(%s)' % ', '.join(
                '%s, %s' % (self._ctx.get_expr_compiled(it1), self._ctx.get_expr_compiled(it2))
                for it1, it2 in zip(expr._keys, expr._values)
            )

        if compiled is not None:
            self._ctx.add_expr_compiled(expr, compiled)
        else:
            raise NotImplementedError

    def visit_groupby(self, expr):
        bys, having, aggs, fields = tuple(expr.args[1:])
        if fields is None:
            fields = bys + aggs

        by_fields = [self._ctx.get_expr_compiled(by) for by in bys]
        group_by_clause = 'GROUP BY {0} '.format(self._join_compiled_fields(by_fields))

        select_fields = [self._compile_select_field(field) for field in fields]
        select_clause = self._join_select_fields(select_fields)

        self.add_select_clause(expr, select_clause)
        self.add_group_by_clause(expr, group_by_clause)

        if having:
            self.add_having_clause(expr, self._ctx.get_expr_compiled(having))

    def visit_mutate(self, expr):
        bys, mutates, fields = tuple(expr.args[1:])
        if fields is None:
            fields = bys + mutates

        select_fields = [self._compile_select_field(field) for field in fields]
        select_clause = self._join_select_fields(select_fields)

        self.add_select_clause(expr, select_clause)

    def visit_sort_column(self, expr):
        def get_field(field):
            if isinstance(field.input, CollectionExpr):
                return field._source_name
            elif isinstance(field.input, Column):
                return field.input.source_name
            else:
                return self._ctx.get_expr_compiled(field.input)
        compiled = '{0} DESC'.format(get_field(expr)) \
            if not expr._ascending else get_field(expr)
        self._ctx.add_expr_compiled(expr, compiled)

    def visit_sort(self, expr):
        keys_fields = expr.args[1]

        order_by_clause = 'ORDER BY {0} '.format(self._join_compiled_fields(
            [self._ctx.get_expr_compiled(field) for field in keys_fields]))

        self.add_order_by_clause(expr, order_by_clause)

    def visit_sample(self, expr):
        assert expr._parts is None and not expr._replace.value
        if expr._frac is not None:
            self.add_table_sample_clause(expr, "({0} PERCENT)".format(int(expr._frac.value * 100)))
        elif expr._n is not None:
            self.add_table_sample_clause(expr, "({0} ROWS)".format(expr._n.value))

    def visit_distinct(self, expr):
        distinct_fields = expr.args[1]

        fields_clause = self._join_select_fields(
            [self._compile_select_field(field) for field in distinct_fields])
        select_clause = 'DISTINCT {0}'.format(fields_clause)

        self.add_select_clause(expr, select_clause)

    def visit_reduction(self, expr):
        is_unique = getattr(expr, '_unique', False)

        if isinstance(expr, (Count, GroupedCount)) and isinstance(expr.input, CollectionExpr):
            compiled = 'COUNT(1)'
            self._ctx.add_expr_compiled(expr, compiled)
            return

        if isinstance(expr, (Std, GroupedStd)):
            if expr._ddof not in (0, 1):
                raise CompileError('Does not support %s with ddof=%s' % (
                    expr.node_name, expr._ddof))

        compiled = None

        if isinstance(expr, (Mean, GroupedMean)):
            node_name = 'avg'
        elif isinstance(expr, (Std, GroupedStd)):
            node_name = 'stddev' if expr._ddof == 0 else 'stddev_samp'
        elif isinstance(expr, (Sum, GroupedSum)) and expr.input.dtype == df_types.string:
            if is_unique:
                compiled = 'WM_CONCAT(DISTINCT \'\', %s)' % self._ctx.get_expr_compiled(expr.input)
            else:
                compiled = 'WM_CONCAT(\'\', %s)' % self._ctx.get_expr_compiled(expr.input)
        elif isinstance(expr, (Sum, GroupedSum)) and expr.input.dtype == df_types.boolean:
            if getattr(expr, '_unique', False):
                compiled = 'SUM(DISTINCT IF(%s, 1, 0))' % self._ctx.get_expr_compiled(expr.input)
            else:
                compiled = 'SUM(IF(%s, 1, 0))' % self._ctx.get_expr_compiled(expr.input)
        elif isinstance(expr, (Max, GroupedMax, Min, GroupedMin)) and \
                expr.input.dtype == df_types.boolean:
            compiled = '%s(IF(%s, 1, 0)) == 1' % (
                expr.node_name, self._ctx.get_expr_compiled(expr.input))
        elif isinstance(expr, (Any, GroupedAny)):
            compiled = 'MAX(IF(%s, 1, 0)) == 1' % self._ctx.get_expr_compiled(expr.args[0])
        elif isinstance(expr, (All, GroupedAll)):
            compiled = 'MIN(IF(%s, 1, 0)) == 1' % self._ctx.get_expr_compiled(expr.args[0])
        elif isinstance(expr, (NUnique, GroupedNUnique)):
            compiled = 'COUNT(DISTINCT %s)' % ', '.join(
                self._ctx.get_expr_compiled(c) for c in expr.inputs)
        elif isinstance(expr, (Cat, GroupedCat)):
            if is_unique:
                compiled = 'WM_CONCAT(DISTINCT %s, %s)' % (self._ctx.get_expr_compiled(expr._sep),
                                                           self._ctx.get_expr_compiled(expr.input))
            else:
                compiled = 'WM_CONCAT(%s, %s)' % (self._ctx.get_expr_compiled(expr._sep),
                                                  self._ctx.get_expr_compiled(expr.input))
        elif isinstance(expr, (Quantile, GroupedQuantile)):
            if not isinstance(expr._prob, (list, set)):
                probs_expr = expr._prob
            else:
                probs_expr = 'ARRAY(' + ', '.join(str(p) for p in expr._prob) + ')'

            if expr.input.data_type in (df_types.float32, df_types.float64) and types.get_local_use_odps2_types():
                func_name = 'PERCENTILE_APPROX'
            else:
                func_name = 'PERCENTILE'

            if is_unique:
                compiled = '%s(DISTINCT %s, %s)' % (func_name, self._ctx.get_expr_compiled(expr.input), probs_expr)
            else:
                compiled = '%s(%s, %s)' % (func_name, self._ctx.get_expr_compiled(expr.input), probs_expr)
        elif isinstance(expr, (ToList, GroupedToList)):
            func_name = 'COLLECT_SET' if expr._unique else 'COLLECT_LIST'
            compiled = '%s(%s)' % (func_name, self._ctx.get_expr_compiled(expr.input))
        else:
            node_name = expr.node_name

        if compiled is None:
            if is_unique:
                compiled = '{0}(DISTINCT {1})'.format(
                    node_name.upper(), self._ctx.get_expr_compiled(expr.args[0]))
            else:
                compiled = '{0}({1})'.format(
                    node_name.upper(), self._ctx.get_expr_compiled(expr.args[0]))

        self._ctx.add_expr_compiled(expr, compiled)

    def visit_user_defined_aggregator(self, expr):
        is_func_created = False
        if isinstance(expr._aggregator, six.string_types):
            func_name = expr._aggregator
        elif isinstance(expr._aggregator, Function):
            func_name = expr._aggregator.name
        else:
            func_name = self._ctx.get_udf(expr._aggregator)
            is_func_created = True

        args = [self._ctx.get_expr_compiled(i) for i in expr.inputs]

        if hasattr(expr, '_func_args') and expr._func_args is not None \
                and not is_func_created:
            func_args = [repr(arg) for arg in expr._func_args]
            args.extend(func_args)
        if getattr(expr, '_unique', False):
            compiled = '{0}(DISTINCT {1})'.format(func_name, ', '.join(args))
        else:
            compiled = '{0}({1})'.format(func_name, ', '.join(args))

        self._ctx.add_expr_compiled(expr, compiled)

    def visit_column(self, expr):
        alias = self._ctx.get_collection_alias(expr.input, silent=True)

        if alias:
            alias = alias[0]
            compiled = '{0}.{1}'.format(alias, self._quote(expr.source_name))
        else:
            symbol = self._ctx.add_need_alias_column(expr)
            compiled = '%({0})s'.format(symbol)

        if expr._source_data_type != expr._data_type:
            compiled = 'CAST({0} AS {1})'.format(
                compiled, types.df_type_to_odps_type(expr.dtype))

        self._ctx.add_expr_compiled(expr, compiled)

    def visit_function(self, expr):
        is_func_created = False
        if isinstance(expr, Func):
            func_name = expr._func_name
        else:
            if isinstance(expr._func, six.string_types):
                func_name = expr._func
            elif isinstance(expr._func, Function):
                func_name = expr._func.name
            else:
                func_name = self._ctx.get_udf(expr._func)
                is_func_created = True

        if isinstance(expr, (MappedExpr, Func)):
            args = [self._ctx.get_expr_compiled(f) for f in expr.inputs]
        else:
            raise NotImplementedError
        if hasattr(expr, '_func_args') and expr._func_args is not None \
                and not is_func_created:
            func_args = [repr(arg) for arg in expr._func_args]
            args.extend(func_args)
        compiled = '{0}({1})'.format(func_name, ', '.join(args))

        self._ctx.add_expr_compiled(expr, compiled)

    def visit_builtin_function(self, expr):
        compiled = '{0}({1})'.format(expr._func_name,
                                     ', '.join(repr(arg) for arg in expr._func_args))
        self._ctx.add_expr_compiled(expr, compiled)

    def _quote(self, compiled):
        if options.df.quote:
            return '`{0}`'.format(compiled)
        else:
            return compiled

    def _unquote(self, compiled):
        if options.df.quote:
            reg = re.compile(r'`([^`]+)`')
        else:
            reg = re.compile(r'\.(\w+)')
        matched = reg.search(compiled)
        if matched:
            return matched.group(1)
        return compiled

    def visit_reshuffle(self, expr):
        bys, sorts = expr._by, expr._sort_fields

        by_fields = [self._unquote(self._ctx.get_expr_compiled(by)) for by in bys]
        distribute_by_clause = 'DISTRIBUTE BY {0} '.format(
            self._join_compiled_fields(by_fields))

        self.add_group_by_clause(expr, distribute_by_clause)

        if sorts:
            sort_fields = [self._ctx.get_expr_compiled(sort) for sort in sorts]
            sort_by_clause = 'SORT BY {0}'.format(self._join_compiled_fields(sort_fields))

            self.add_order_by_clause(expr, sort_by_clause)

    def visit_apply_collection(self, expr):
        is_func_created = False
        if isinstance(expr._func, six.string_types):
            func_name = expr._func
        elif isinstance(expr._func, Function):
            func_name = expr._func.name
        else:
            func_name = self._ctx.get_udf(expr._func)
            is_func_created = True

        args = [self._ctx.get_expr_compiled(f) for f in expr._fields]
        if hasattr(expr, '_func_args') and expr._func_args is not None \
                and not is_func_created:
            func_args = [repr(arg) for arg in expr._func_args]
            args.extend(func_args)

        func_call = '{0}({1})'.format(func_name, ', '.join(args))
        field_list = ', '.join(self._quote(n) for n in expr.schema.names)
        if expr._lateral_view:
            lv_prefix = 'LATERAL VIEW ' if not expr._keep_nulls else 'LATERAL VIEW OUTER '
            compiled = lv_prefix + '{0} {1} AS {2}'.format(
                func_call, self._ctx.get_collection_alias(expr, create=True)[0], field_list)
        else:
            compiled = '{0} AS ({1})'.format(func_call, field_list)
            self.add_select_clause(expr, compiled)

        self._ctx.add_expr_compiled(expr, compiled)

    def _wrap_typed(self, expr, compiled):
        if expr._source_data_type != expr._data_type:
            compiled = 'cast({0} AS {1})'.format(
                compiled, types.df_type_to_odps_type(expr._data_type))

        return compiled

    def visit_sequence(self, expr):
        compiled = expr._source_name
        compiled = self._wrap_typed(expr, compiled)

        self._ctx.add_expr_compiled(expr, compiled)

    def _compile_window_order_by(self, expr):
        if isinstance(expr.input, SequenceExpr):
            compiled = self._ctx.get_expr_compiled(expr.input)
            return '%s DESC' % compiled if not expr._ascending else compiled
        else:
            return self._ctx.get_expr_compiled(expr)

    def _compile_window_function(self, func, args, partition_by=None,
                                 order_by=None, preceding=None, following=None):
        partition_by = 'PARTITION BY {0}'.format(partition_by or '1')
        order_by = 'ORDER BY {0}'.format(order_by) if order_by is not None else ''

        if isinstance(preceding, tuple):
            window_clause = 'ROWS BETWEEN {0} PRECEDING AND {1} PRECEDING' \
                .format(*[self._ctx.get_expr_compiled(p) for p in preceding])
        elif isinstance(following, tuple):
            window_clause = 'ROWS BETWEEN {0} FOLLOWING AND {1} FOLLOWING' \
                .format(*[self._ctx.get_expr_compiled(f) for f in following])
        elif preceding is not None and following is not None:
            window_clause = 'ROWS BETWEEN {0} PRECEDING AND {1} FOLLOWING' \
                .format(self._ctx.get_expr_compiled(preceding), self._ctx.get_expr_compiled(following))
        elif preceding is not None:
            window_clause = 'ROWS {0} PRECEDING'.format(self._ctx.get_expr_compiled(preceding))
        elif following is not None:
            window_clause = 'ROWS {0} FOLLOWING'.format(self._ctx.get_expr_compiled(following))
        else:
            window_clause = ''

        over = ' '.join(sub for sub in (partition_by, order_by, window_clause)
                        if len(sub) > 0)

        return '{0}({1}) OVER ({2})'.format(func, args, over)

    def visit_cum_window(self, expr):
        col_compiled = self._ctx.get_expr_compiled(expr.input)
        if isinstance(expr, CumSum) and expr.input.dtype == df_types.boolean:
            col_compiled = 'IF({0}, 1, 0)'.format(col_compiled)
        elif isinstance(expr, NthValue):
            col_compiled = '{0}, {1}'.format(col_compiled, expr._nth + 1)
            if expr._skip_nulls:
                col_compiled += ', true'
        if expr.distinct:
            col_compiled = 'DISTINCT {0}'.format(col_compiled)

        partition_by = ', '.join(self._ctx.get_expr_compiled(by)
                                 for by in expr._partition_by) if expr._partition_by else None
        order_by = ', '.join(self._compile_window_order_by(by)
                             for by in expr._order_by) if expr._order_by else None

        func_name = WINDOW_COMPILE_DIC[expr.node_name].upper()
        compiled = self._compile_window_function(func_name, col_compiled, partition_by=partition_by,
                                                 order_by=order_by, preceding=expr._preceding,
                                                 following=expr._following)

        self._ctx.add_expr_compiled(expr, compiled)

    def visit_rank_window(self, expr):
        func_name = WINDOW_COMPILE_DIC[expr.node_name].upper()
        col_compiled = ''
        if isinstance(expr, QCut):
            col_compiled = str(expr._bins)

        partition_by = ', '.join(self._ctx.get_expr_compiled(by)
                                 for by in expr._partition_by) if expr._partition_by else None
        order_by = ', '.join(self._compile_window_order_by(by)
                             for by in expr._order_by) if expr._order_by else None

        compiled = self._compile_window_function(func_name, col_compiled, partition_by=partition_by,
                                                 order_by=order_by)

        self._ctx.add_expr_compiled(expr, compiled)

    def visit_shift_window(self, expr):
        func_name = WINDOW_COMPILE_DIC[expr.node_name].upper()

        compiled_fields = [self._ctx.get_expr_compiled(expr.input), ]
        if expr._offset:
            compiled_fields.append(self._ctx.get_expr_compiled(expr._offset))
        if expr._default:
            compiled_fields.append(self._ctx.get_expr_compiled(expr._default))

        col_compiled = self._join_compiled_fields(compiled_fields)

        partition_by = ', '.join(self._ctx.get_expr_compiled(by)
                                 for by in expr._partition_by) if expr._partition_by else None
        order_by = ', '.join(self._compile_window_order_by(by)
                             for by in expr._order_by) if expr._order_by else None

        compiled = self._compile_window_function(func_name, col_compiled, partition_by=partition_by,
                                                 order_by=order_by)

        self._ctx.add_expr_compiled(expr, compiled)

    def visit_scalar(self, expr):
        compiled = None
        if expr._value is not None:
            if expr.dtype == df_types.string:
                val = utils.to_str(expr.value) \
                    if isinstance(expr.value, six.text_type) else expr.value
                compiled = "'{0}'".format(val.replace("'", "\\'"))
            elif isinstance(expr.dtype, df_types.Integer) and types.get_local_use_odps2_types():
                if expr.dtype == df_types.int8:
                    compiled = "{0!r}Y".format(expr._value)
                elif expr.dtype == df_types.int16:
                    compiled = "{0!r}S".format(expr._value)
                elif expr.dtype == df_types.int64:
                    compiled = "{0!r}L".format(expr._value)
            elif isinstance(expr._value, bool):
                compiled = 'true' if expr._value else 'false'
            elif isinstance(expr._value, datetime):
                # FIXME: just ignore shorter than second
                compiled = 'FROM_UNIXTIME({0})'.format(utils.to_timestamp(expr._value))
            elif isinstance(expr._value, date):
                compiled = 'CAST({0!r} AS DATE)'.format(expr._value.strftime("%Y-%m-%d"))
            elif isinstance(expr._value, Decimal):
                compiled = 'CAST({0} AS DECIMAL)'.format(repr(str(expr._value)))
        else:
            compiled = 'CAST(NULL AS {0})'.format(types.df_type_to_odps_type(expr._value_type))

        if compiled is None:
            compiled = repr(expr._value)
        self._ctx.add_expr_compiled(expr, compiled)

    @classmethod
    def _cast(cls, compiled, source_type, to_type):
        source_odps_type = types.df_type_to_odps_type(source_type)
        to_type = types.df_type_to_odps_type(to_type)

        if not to_type.can_explicit_cast(source_odps_type):
            raise CompileError(
                    'Cannot cast from %s to %s' % (source_odps_type, to_type))

        return 'CAST({0} AS {1})'.format(compiled, to_type)

    def visit_cast(self, expr):
        compiled = self._ctx.get_expr_compiled(expr._input)

        if isinstance(expr.source_type, df_types.Integer) and expr.dtype == df_types.datetime:
            compiled = 'FROM_UNIXTIME({0})'.format(self._ctx.get_expr_compiled(expr.input))
        elif expr.dtype != expr.source_type:
            compiled = self._cast(compiled, expr.source_type, expr.dtype)

        self._ctx.add_expr_compiled(expr, compiled)

    def visit_join(self, expr):
        left_compiled, right_compiled, predicate_compiled = tuple(self._sub_compiles[expr])

        from_clause = '{0} \n{1} JOIN \n{2}'.format(
            left_compiled, expr._how, utils.indent(right_compiled, self._indent_size)
        )
        if predicate_compiled:
            from_clause += '\nON {0}'.format(predicate_compiled)

        self.add_from_clause(expr, from_clause)
        self._ctx.add_expr_compiled(expr, from_clause)

    def visit_union(self, expr):
        union_type = 'UNION ALL' if not expr._distinct else 'UNION'
        left_compiled, right_compiled = tuple(self._sub_compiles[expr])

        from_clause = '{0} \n{1}\n{2}'.format(left_compiled, union_type,
                                              utils.indent(right_compiled, self._indent_size))

        compiled = from_clause
        if not self._union_no_alias.get(expr, False):
            compiled = '(\n{0}\n) {1}'.format(utils.indent(from_clause, self._indent_size),
                                              self._ctx.get_collection_alias(expr, create=True)[0])

        self.add_from_clause(expr, compiled)
        self._ctx.add_expr_compiled(expr, compiled)