[FLINK-11449][table] Uncouple table expressions from Calcite

This commit uncouples table expressions from Calcite by switching
the API to a new, simplified basic expression stack.

Major changes:
- Rename CommonExpression to Expression
- Rework the function catalog to work with the newly introduced function definitions
- Rework the Scala implicits and Java expression parser to return the new expression stack
- Introduce an expression bridge on the edges of the API to convert to the old expression stack as a temporary solution until the old expression stack becomes obsolete
- Cleanup of main API interfaces (i.e. Table.scala and expressionDsl.scala)
- Ensure that SPI classes such as FilterableTableSource and TimestampExtractor still work

This closes apache#7664.
This closes apache#7967.

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/AggregateFunctionDefinition.java

@@ -36,10 +36,11 @@ public final class AggregateFunctionDefinition extends FunctionDefinition {
  private final TypeInformation<?> accumulatorTypeInfo;
 
  public AggregateFunctionDefinition(
+ String name,
  AggregateFunction<?, ?> aggregateFunction,
  TypeInformation<?> resultTypeInfo,
  TypeInformation<?> accTypeInfo) {
- super(aggregateFunction.getClass().getName(), AGGREGATE_FUNCTION);
+ super(name, AGGREGATE_FUNCTION);
  this.aggregateFunction = Preconditions.checkNotNull(aggregateFunction);
  this.resultTypeInfo = Preconditions.checkNotNull(resultTypeInfo);
  this.accumulatorTypeInfo = Preconditions.checkNotNull(accTypeInfo);

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/BuiltInFunctionDefinitions.java

@@ -44,7 +44,7 @@ public final class BuiltInFunctionDefinitions {
  public static final FunctionDefinition NOT =
  new FunctionDefinition("not", SCALAR_FUNCTION);
  public static final FunctionDefinition IF =
- new FunctionDefinition("if", SCALAR_FUNCTION);
+ new FunctionDefinition("ifThenElse", SCALAR_FUNCTION);
 
  // comparison functions
  public static final FunctionDefinition EQUALS =
@@ -110,7 +110,7 @@ public final class BuiltInFunctionDefinitions {
  public static final FunctionDefinition LIKE =
  new FunctionDefinition("like", SCALAR_FUNCTION);
  public static final FunctionDefinition LOWER =
- new FunctionDefinition("lower", SCALAR_FUNCTION);
+ new FunctionDefinition("lowerCase", SCALAR_FUNCTION);
  public static final FunctionDefinition SIMILAR =
  new FunctionDefinition("similar", SCALAR_FUNCTION);
  public static final FunctionDefinition SUBSTRING =
@@ -120,7 +120,7 @@ public final class BuiltInFunctionDefinitions {
  public static final FunctionDefinition TRIM =
  new FunctionDefinition("trim", SCALAR_FUNCTION);
  public static final FunctionDefinition UPPER =
- new FunctionDefinition("upper", SCALAR_FUNCTION);
+ new FunctionDefinition("upperCase", SCALAR_FUNCTION);
  public static final FunctionDefinition POSITION =
  new FunctionDefinition("position", SCALAR_FUNCTION);
  public static final FunctionDefinition OVERLAY =
@@ -249,10 +249,6 @@ public final class BuiltInFunctionDefinitions {
  new FunctionDefinition("dateFormat", SCALAR_FUNCTION);
  public static final FunctionDefinition TIMESTAMP_DIFF =
  new FunctionDefinition("timestampDiff", SCALAR_FUNCTION);
- public static final FunctionDefinition TEMPORAL_FLOOR =
- new FunctionDefinition("temporalFloor", SCALAR_FUNCTION);
- public static final FunctionDefinition TEMPORAL_CEIL =
- new FunctionDefinition("temporalCeil", SCALAR_FUNCTION);
 
  // collection
  public static final FunctionDefinition AT =

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/CallExpression.java

@@ -33,19 +33,19 @@
  * <p>The function can be a built-in function or a user-defined function.
  */
 @PublicEvolving
-public final class CallExpression implements CommonExpression {
+public final class CallExpression implements Expression {
 
  private final FunctionDefinition functionDefinition;
 
- private final List<CommonExpression> args;
+ private final List<Expression> args;
 
- public CallExpression(FunctionDefinition functionDefinition, List<CommonExpression> args) {
+ public CallExpression(FunctionDefinition functionDefinition, List<Expression> args) {
  this.functionDefinition = Preconditions.checkNotNull(functionDefinition);
  this.args = Collections.unmodifiableList(new ArrayList<>(Preconditions.checkNotNull(args)));
  }
 
  @Override
- public List<CommonExpression> getChildren() {
+ public List<Expression> getChildren() {
  return this.args;
  }
 

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/CommonExpression.java → flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/Expression.java

@@ -30,9 +30,9 @@
  * or field references.
  */
 @PublicEvolving
-public interface CommonExpression {
+public interface Expression {
 
- List<CommonExpression> getChildren();
+ List<Expression> getChildren();
 
  <R> R accept(ExpressionVisitor<R> visitor);
 }

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java

@@ -21,22 +21,22 @@
 import org.apache.flink.annotation.PublicEvolving;
 
 /**
- * The visitor definition of {@link CommonExpression}. An expression visitor transforms an
+ * The visitor definition of {@link Expression}. An expression visitor transforms an
  * expression to instances of {@code R}.
  */
 @PublicEvolving
 public interface ExpressionVisitor<R> {
 
  R visitCall(CallExpression call);
 
- R visitSymbol(CommonSymbolExpression symbolExpression);
+ R visitSymbol(SymbolExpression symbolExpression);
 
  R visitValueLiteral(ValueLiteralExpression valueLiteralExpression);
 
  R visitFieldReference(FieldReferenceExpression fieldReference);
 
  R visitTypeLiteral(TypeLiteralExpression typeLiteral);
 
- R visit(CommonExpression other);
+ R visit(Expression other);
 
 }

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/FieldReferenceExpression.java

@@ -35,7 +35,7 @@
  * References that come out of the optimizer are likely to be resolved and typed.
  */
 @PublicEvolving
-public final class FieldReferenceExpression implements CommonExpression {
+public final class FieldReferenceExpression implements Expression {
 
  private final String name;
 
@@ -60,7 +60,7 @@ public Optional<TypeInformation<?>> getResultType() {
  }
 
  @Override
- public List<CommonExpression> getChildren() {
+ public List<Expression> getChildren() {
  return Collections.emptyList();
  }
 

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ScalarFunctionDefinition.java

@@ -32,8 +32,8 @@ public final class ScalarFunctionDefinition extends FunctionDefinition {
 
  private final ScalarFunction scalarFunction;
 
- public ScalarFunctionDefinition(ScalarFunction scalarFunction) {
- super(scalarFunction.getClass().getName(), SCALAR_FUNCTION);
+ public ScalarFunctionDefinition(String name, ScalarFunction scalarFunction) {
+ super(name, SCALAR_FUNCTION);
  this.scalarFunction = Preconditions.checkNotNull(scalarFunction);
  }
 

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/CommonSymbolExpression.java → flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/SymbolExpression.java

@@ -29,20 +29,20 @@
  * An expression that wraps a specific symbol.
  */
 @PublicEvolving
-public final class CommonSymbolExpression implements CommonExpression {
+public final class SymbolExpression implements Expression {
 
- private final CommonTableSymbol symbol;
+ private final TableSymbol symbol;
 
- public CommonSymbolExpression(CommonTableSymbol symbol) {
+ public SymbolExpression(TableSymbol symbol) {
  this.symbol = Preconditions.checkNotNull(symbol);
  }
 
- public CommonTableSymbol getSymbol() {
+ public TableSymbol getSymbol() {
  return symbol;
  }
 
  @Override
- public List<CommonExpression> getChildren() {
+ public List<Expression> getChildren() {
  return Collections.emptyList();
  }
 
@@ -59,7 +59,7 @@ public boolean equals(Object o) {
  if (o == null || getClass() != o.getClass()) {
  return false;
  }
- CommonSymbolExpression that = (CommonSymbolExpression) o;
+ SymbolExpression that = (SymbolExpression) o;
  return Objects.equals(symbol, that.symbol);
  }
 

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/TableFunctionDefinition.java

@@ -34,8 +34,11 @@ public final class TableFunctionDefinition extends FunctionDefinition {
  private final TableFunction tableFunction;
  private final TypeInformation resultType;
 
- public TableFunctionDefinition(TableFunction tableFunction, TypeInformation resultType) {
- super(tableFunction.getClass().getName(), TABLE_FUNCTION);
+ public TableFunctionDefinition(
+ String name,
+ TableFunction tableFunction,
+ TypeInformation resultType) {
+ super(name, TABLE_FUNCTION);
  this.tableFunction = Preconditions.checkNotNull(tableFunction);
  this.resultType = Preconditions.checkNotNull(resultType);
  }

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/CommonTableSymbol.java → flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/TableSymbol.java

@@ -24,5 +24,5 @@
  * The base interface for all table symbols. Symbols might be time units or other enumerations.
  */
 @PublicEvolving
-public interface CommonTableSymbol {
+public interface TableSymbol {
 }

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/CommonTimeIntervalUnit.java → flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/TimeIntervalUnit.java

@@ -24,7 +24,7 @@
  * Units for working with time intervals.
  */
 @PublicEvolving
-public enum CommonTimeIntervalUnit implements CommonTableSymbol {
+public enum TimeIntervalUnit implements TableSymbol {
  YEAR,
  YEAR_TO_MONTH,
  QUARTER,

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/CommonTimePointUnit.java → flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/TimePointUnit.java

@@ -24,7 +24,7 @@
  * Units for working with points in time.
  */
 @PublicEvolving
-public enum CommonTimePointUnit implements CommonTableSymbol {
+public enum TimePointUnit implements TableSymbol {
  YEAR,
  MONTH,
  DAY,

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/TypeLiteralExpression.java

@@ -31,7 +31,7 @@
  * Expression that wraps {@link TypeInformation} as a literal.
  */
 @PublicEvolving
-public final class TypeLiteralExpression implements CommonExpression {
+public final class TypeLiteralExpression implements Expression {
 
  private final TypeInformation<?> type;
 
@@ -44,7 +44,7 @@ public TypeInformation<?> getType() {
  }
 
  @Override
- public List<CommonExpression> getChildren() {
+ public List<Expression> getChildren() {
  return Collections.emptyList();
  }
 

flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ValueLiteralExpression.java

@@ -38,7 +38,7 @@
  * Expression for constant literal values.
  */
 @PublicEvolving
-public final class ValueLiteralExpression implements CommonExpression {
+public final class ValueLiteralExpression implements Expression {
 
  private final Object value;
 
@@ -67,7 +67,7 @@ public TypeInformation<?> getType() {
  }
 
  @Override
- public List<CommonExpression> getChildren() {
+ public List<Expression> getChildren() {
  return Collections.emptyList();
  }
 

flink-table/flink-table-common/src/test/java/org/apache/flink/table/expressions/ExpressionTest.java

@@ -31,22 +31,22 @@
 import static org.junit.Assert.assertNotEquals;
 
 /**
- * Tests for {@link org.apache.flink.table.expressions.CommonExpression} and its sub-classes.
+ * Tests for {@link org.apache.flink.table.expressions.Expression} and its sub-classes.
  */
 public class ExpressionTest {
 
  private static final ScalarFunction DUMMY_FUNCTION = new ScalarFunction() {
  // dummy
  };
 
- private static final CommonExpression TREE_WITH_NULL = createExpressionTree(null);
+ private static final Expression TREE_WITH_NULL = createExpressionTree(null);
 
- private static final CommonExpression TREE_WITH_VALUE = createExpressionTree(12);
+ private static final Expression TREE_WITH_VALUE = createExpressionTree(12);
 
- private static final CommonExpression TREE_WITH_SAME_VALUE = createExpressionTree(12);
+ private static final Expression TREE_WITH_SAME_VALUE = createExpressionTree(12);
 
  private static final String TREE_WITH_NULL_STRING =
- "and(true, equals(field, " + ExpressionTest.class.getName() + "$1(null)))";
+ "and(true, equals(field, dummy(null)))";
 
  @Test
  public void testExpressionString() {
@@ -63,7 +63,7 @@ public void testExpressionInequality() {
  assertNotEquals(TREE_WITH_NULL, TREE_WITH_VALUE);
  }
 
- private static CommonExpression createExpressionTree(Integer nestedValue) {
+ private static Expression createExpressionTree(Integer nestedValue) {
  return new CallExpression(
  AND,
  asList(
@@ -73,7 +73,7 @@ private static CommonExpression createExpressionTree(Integer nestedValue) {
  asList(
  new FieldReferenceExpression("field"),
  new CallExpression(
- new ScalarFunctionDefinition(DUMMY_FUNCTION),
+ new ScalarFunctionDefinition("dummy", DUMMY_FUNCTION),
  singletonList(new ValueLiteralExpression(nestedValue, Types.INT))
  )
  )