[FLINK-21808][table-api-java] Add NOP operation

This closes apache#15253

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java

@@ -91,6 +91,7 @@
 import org.apache.flink.table.operations.ExplainOperation;
 import org.apache.flink.table.operations.LoadModuleOperation;
 import org.apache.flink.table.operations.ModifyOperation;
+import org.apache.flink.table.operations.NopOperation;
 import org.apache.flink.table.operations.Operation;
 import org.apache.flink.table.operations.QueryOperation;
 import org.apache.flink.table.operations.SelectSinkOperation;
@@ -673,12 +674,22 @@ public String explainSql(String statement, ExplainDetail... extraDetails) {
  "Unsupported SQL query! explainSql() only accepts a single SQL query.");
  }
 
- return planner.explain(operations, extraDetails);
+ return explainInternal(operations, extraDetails);
  }
 
  @Override
  public String explainInternal(List<Operation> operations, ExplainDetail... extraDetails) {
- return planner.explain(operations, extraDetails);
+ operations =
+ operations.stream()
+ .filter(o -> !(o instanceof NopOperation))
+ .collect(Collectors.toList());
+ // hive parser may generate an NopOperation, in which case we just return an
+ // empty string as the plan
+ if (operations.isEmpty()) {
+ return "";
+ } else {
+ return planner.explain(operations, extraDetails);
+ }
  }
 
  @Override
@@ -821,7 +832,8 @@ public void sqlUpdate(String stmt) {
  || operation instanceof UseCatalogOperation
  || operation instanceof UseDatabaseOperation
  || operation instanceof LoadModuleOperation
- || operation instanceof UnloadModuleOperation) {
+ || operation instanceof UnloadModuleOperation
+ || operation instanceof NopOperation) {
  executeInternal(operation);
  } else {
  throw new TableException(UNSUPPORTED_QUERY_IN_SQL_UPDATE_MSG);
@@ -1178,7 +1190,7 @@ public TableResult executeInternal(Operation operation) {
  }
  } else if (operation instanceof ExplainOperation) {
  String explanation =
- planner.explain(
+ explainInternal(
  Collections.singletonList(((ExplainOperation) operation).getChild()));
  return TableResultImpl.builder()
  .resultKind(ResultKind.SUCCESS_WITH_CONTENT)
@@ -1203,6 +1215,8 @@ public TableResult executeInternal(Operation operation) {
  } else if (operation instanceof CreateTableASOperation) {
  executeInternal(((CreateTableASOperation) operation).getCreateTableOperation());
  return executeInternal(((CreateTableASOperation) operation).getInsertOperation());
+ } else if (operation instanceof NopOperation) {
+ return TableResultImpl.TABLE_RESULT_OK;
  } else {
  throw new TableException(UNSUPPORTED_QUERY_IN_EXECUTE_SQL_MSG);
  }

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/NopOperation.java

@@ -0,0 +1,28 @@
+/*
+ * 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:https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.operations;
+
+/** An {@link Operation} to represent that nothing needs to be done. */
+public class NopOperation implements Operation {
+
+ @Override
+ public String asSummaryString() {
+ return "NOP";
+ }
+}