[FLINK-21886][table-api] Loosen argument of TableEnvironmentInternal#…

…executeInternal from QueryOperation to Operation (apache#15296)
deepfield · Mar 21, 2021 · d044e5c · d044e5c
1 parent 10f1962
commit d044e5c
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 16 deletions.
diff --git a/...able-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/...able-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -708,7 +708,7 @@ public TableResult executeSql(String statement) {
  throw new TableException(UNSUPPORTED_QUERY_IN_EXECUTE_SQL_MSG);
  }
 
- return executeOperation(operations.get(0));
+ return executeInternal(operations.get(0));
  }
 
  @Override
@@ -750,8 +750,7 @@ private TableResult executeInternal(
  }
  }
 
- @Override
- public TableResult executeInternal(QueryOperation operation) {
+ private TableResult executeQueryOperation(QueryOperation operation) {
  SelectSinkOperation sinkOperation = new SelectSinkOperation(operation);
  List<Transformation<?>> transformations =
  translate(Collections.singletonList(sinkOperation));
@@ -808,13 +807,14 @@ public void sqlUpdate(String stmt) {
  || operation instanceof UseDatabaseOperation
  || operation instanceof LoadModuleOperation
  || operation instanceof UnloadModuleOperation) {
- executeOperation(operation);
+ executeInternal(operation);
  } else {
  throw new TableException(UNSUPPORTED_QUERY_IN_SQL_UPDATE_MSG);
  }
  }
 
- private TableResult executeOperation(Operation operation) {
+ @Override
+ public TableResult executeInternal(Operation operation) {
  if (operation instanceof ModifyOperation) {
  return executeInternal(Collections.singletonList((ModifyOperation) operation));
  } else if (operation instanceof CreateTableOperation) {
@@ -1184,7 +1184,7 @@ private TableResult executeOperation(Operation operation) {
  describeTableOperation.getSqlIdentifier().asSummaryString()));
  }
  } else if (operation instanceof QueryOperation) {
- return executeInternal((QueryOperation) operation);
+ return executeQueryOperation((QueryOperation) operation);
  } else {
  throw new TableException(UNSUPPORTED_QUERY_IN_EXECUTE_SQL_MSG);
  }

diff --git a/...-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentInternal.java b/...-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentInternal.java
@@ -27,7 +27,6 @@
 import org.apache.flink.table.delegation.Parser;
 import org.apache.flink.table.operations.ModifyOperation;
 import org.apache.flink.table.operations.Operation;
-import org.apache.flink.table.operations.QueryOperation;
 import org.apache.flink.table.sinks.TableSink;
 import org.apache.flink.table.sources.TableSource;
 
@@ -62,12 +61,12 @@ public interface TableEnvironmentInternal extends TableEnvironment {
  TableResult executeInternal(List<ModifyOperation> operations);
 
  /**
- * Execute the given query operation and return the execution result.
+ * Execute the given operation and return the execution result.
  *
- * @param operation The QueryOperation to be executed.
- * @return the content of the QueryOperation.
+ * @param operation The operation to be executed.
+ * @return the content of the execution result.
  */
- TableResult executeInternal(QueryOperation operation);
+ TableResult executeInternal(Operation operation);
 
  /**
  * Returns the AST of this table and the execution plan to compute the result of this table.

diff --git a/...flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/TableEnvImpl.scala b/...flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/TableEnvImpl.scala
@@ -568,7 +568,7 @@ abstract class TableEnvImpl(
  if (operations.size != 1) {
  throw new TableException(UNSUPPORTED_QUERY_IN_EXECUTE_SQL_MSG)
  }
- executeOperation(operations.get(0))
+ executeInternal(operations.get(0))
  }
 
  override def createStatementSet = new StatementSetImpl(this)
@@ -609,7 +609,7 @@ abstract class TableEnvImpl(
  }
  }
 
- override def executeInternal(operation: QueryOperation): TableResult = {
+ private def executeQueryOperation(operation: QueryOperation): TableResult = {
  val tableSchema = TableSchema.fromResolvedSchema(operation.getResolvedSchema)
  val tableSink = new BatchSelectTableSink(tableSchema)
  val dataSink = writeToSinkAndTranslate(operation, tableSink)
@@ -651,12 +651,12 @@ abstract class TableEnvImpl(
  _: CreateCatalogFunctionOperation | _: CreateTempSystemFunctionOperation |
  _: DropCatalogFunctionOperation | _: DropTempSystemFunctionOperation |
  _: AlterCatalogFunctionOperation | _: UseCatalogOperation | _: UseDatabaseOperation =>
- executeOperation(operation)
+ executeInternal(operation)
  case _ => throw new TableException(UNSUPPORTED_QUERY_IN_SQL_UPDATE_MSG)
  }
  }
 
- private def executeOperation(operation: Operation): TableResult = {
+ override def executeInternal(operation: Operation): TableResult = {
  operation match {
  case catalogSinkModifyOperation: CatalogSinkModifyOperation =>
  executeInternal(JCollections.singletonList[ModifyOperation](catalogSinkModifyOperation))
@@ -828,7 +828,7 @@ abstract class TableEnvImpl(
  descOperation.getSqlIdentifier.asSummaryString()))
  }
  case queryOperation: QueryOperation =>
- executeInternal(queryOperation)
+ executeQueryOperation(queryOperation)
 
  case _ =>
  throw new TableException(UNSUPPORTED_QUERY_IN_EXECUTE_SQL_MSG)