[FLINK-21741][sql-client] Support SHOW JARS statement in SQL Client (a…

…pache#16010)

flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java

@@ -43,6 +43,7 @@
 import org.apache.flink.table.operations.command.QuitOperation;
 import org.apache.flink.table.operations.command.ResetOperation;
 import org.apache.flink.table.operations.command.SetOperation;
+import org.apache.flink.table.operations.command.ShowJarsOperation;
 import org.apache.flink.table.operations.ddl.AlterOperation;
 import org.apache.flink.table.operations.ddl.CreateOperation;
 import org.apache.flink.table.operations.ddl.DropOperation;
@@ -425,6 +426,9 @@ private void callOperation(Operation operation, ExecutionMode mode) {
  } else if (operation instanceof AddJarOperation) {
  // ADD JAR
  callAddJar((AddJarOperation) operation);
+ } else if (operation instanceof ShowJarsOperation) {
+ // SHOW JARS
+ callShowJars();
  } else if (operation instanceof ShowCreateTableOperation) {
  // SHOW CREATE TABLE
  callShowCreateTable((ShowCreateTableOperation) operation);
@@ -440,6 +444,10 @@ private void callAddJar(AddJarOperation operation) {
  printInfo(CliStrings.MESSAGE_ADD_JAR_STATEMENT);
  }
 
+ private void callShowJars() {
+ executor.listJars(sessionId).forEach(jar -> terminal.writer().println(jar));
+ }
+
  private void callQuit() {
  printInfo(CliStrings.MESSAGE_QUIT);
  isRunning = false;

flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java

@@ -142,4 +142,7 @@ TypedResult<Integer> snapshotResult(String sessionId, String resultId, int pageS
 
  /** Add the JAR resource to into the classloader with specified session. */
  void addJar(String sessionId, String jarPath);
+
+ /** List the JAR resources of the classloader with specified session. */
+ List<String> listJars(String sessionId);
 }

flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/SessionContext.java

@@ -51,8 +51,10 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * Context describing a session, it's mainly used for user to open a new session in the backend. If
@@ -282,6 +284,10 @@ public void addJar(String jarPath) {
  executionContext = new ExecutionContext(sessionConfiguration, classLoader, sessionState);
  }
 
+ public List<String> listJars() {
+ return dependencies.stream().map(URL::getPath).collect(Collectors.toList());
+ }
+
  // --------------------------------------------------------------------------------------------
  // Inner class
  // --------------------------------------------------------------------------------------------

flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java

@@ -313,4 +313,10 @@ public void addJar(String sessionId, String jarUrl) {
  final SessionContext context = getSessionContext(sessionId);
  context.addJar(jarUrl);
  }
+
+ @Override
+ public List<String> listJars(String sessionId) {
+ final SessionContext context = getSessionContext(sessionId);
+ return context.listJars();
+ }
 }

flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java

@@ -523,5 +523,10 @@ public void cancelQuery(String sessionId, String resultId) throws SqlExecutionEx
  public void addJar(String sessionId, String jarUrl) {
  throw new UnsupportedOperationException("Not implemented.");
  }
+
+ @Override
+ public List<String> listJars(String sessionId) {
+ throw new UnsupportedOperationException("Not implemented.");
+ }
  }
 }

flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliResultViewTest.java

@@ -231,6 +231,11 @@ public void cancelQuery(String sessionId, String resultId) throws SqlExecutionEx
  public void addJar(String sessionId, String jarUrl) {
  throw new UnsupportedOperationException("Not implemented.");
  }
+
+ @Override
+ public List<String> listJars(String sessionId) {
+ throw new UnsupportedOperationException("Not implemented.");
+ }
  }
 
  private static final class TestingCliResultView implements Runnable {

flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutor.java

@@ -72,6 +72,11 @@ public void addJar(String sessionId, String jarUrl) {
  throw new UnsupportedOperationException("Not implemented.");
  }
 
+ @Override
+ public List<String> listJars(String sessionId) {
+ throw new UnsupportedOperationException("Not implemented.");
+ }
+
  @Override
  public TypedResult<List<Row>> retrieveResultChanges(String sessionId, String resultId)
  throws SqlExecutionException {

flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/context/SessionContextTest.java

@@ -261,11 +261,13 @@ private Configuration getConfiguration() {
 
  private void validateAddJar(String jarPath) throws IOException {
  sessionContext.addJar(jarPath);
+ assertEquals(Collections.singletonList(udfJar.getPath()), sessionContext.listJars());
  assertEquals(
  Collections.singletonList(udfJar.toURI().toURL().toString()),
  getConfiguration().get(JARS));
  // reset to the default
  sessionContext.reset();
+ assertEquals(Collections.singletonList(udfJar.getPath()), sessionContext.listJars());
  assertEquals(
  Collections.singletonList(udfJar.toURI().toURL().toString()),
  getConfiguration().get(JARS));

flink-table/flink-sql-client/src/test/resources/sql/function.q

@@ -19,6 +19,10 @@ ADD JAR '$VAR_UDF_JAR_PATH';
 [INFO] The specified jar is added into session classloader.
 !info
 
+SHOW JARS;
+$VAR_UDF_JAR_PATH
+!ok
+
 # this also tests user classloader because the LowerUDF is in user jar
 create function func1 as 'LowerUDF' LANGUAGE JAVA;
 [INFO] Execute statement succeed.

flink-table/flink-sql-client/src/test/resources/sql/set.q

@@ -176,6 +176,10 @@ ADD JAR '$VAR_UDF_JAR_PATH';
 [INFO] The specified jar is added into session classloader.
 !info
 
+SHOW JARS;
+$VAR_UDF_JAR_PATH
+!ok
+
 set;
 execution.attached=true
 execution.savepoint.ignore-unclaimed-state=false
@@ -191,6 +195,10 @@ reset;
 [INFO] All session properties have been set to their default values.
 !info
 
+SHOW JARS;
+$VAR_UDF_JAR_PATH
+!ok
+
 SET sql-client.execution.result-mode=tableau;
 [INFO] Session property has been set.
 !info

flink-table/flink-sql-parser-hive/src/main/codegen/data/Parser.tdd

@@ -86,6 +86,7 @@
  "org.apache.flink.sql.parser.dql.SqlShowDatabases"
  "org.apache.flink.sql.parser.dql.SqlShowCurrentDatabase"
  "org.apache.flink.sql.parser.dql.SqlShowFunctions"
+ "org.apache.flink.sql.parser.dql.SqlShowJars"
  "org.apache.flink.sql.parser.dql.SqlShowModules"
  "org.apache.flink.sql.parser.dql.SqlShowTables"
  "org.apache.flink.sql.parser.dql.SqlShowPartitions"
@@ -130,6 +131,7 @@
  "INPUTFORMAT"
  "ITEMS"
  "JAR"
+ "JARS"
  "KEYS"
  "LINES"
  "LOAD"
@@ -272,6 +274,7 @@
  "ISOLATION"
  "ISOYEAR"
  "JAR"
+ "JARS"
  "JAVA"
  "JSON"
  "K"
@@ -543,6 +546,7 @@
  "SqlUseModules()"
  "SqlRichExplain()"
  "SqlAddJar()"
+ "SqlShowJars()"
  ]
 
  # List of methods for parsing custom literals.

flink-table/flink-sql-parser-hive/src/main/codegen/includes/parserImpls.ftl

@@ -1642,3 +1642,17 @@ SqlAddJar SqlAddJar() :
  return new SqlAddJar(getPos(), jarPath);
  }
 }
+
+/**
+* Parses a show jars statement.
+* SHOW JARS;
+*/
+SqlShowJars SqlShowJars() :
+{
+}
+{
+ <SHOW> <JARS>
+ {
+ return new SqlShowJars(getPos());
+ }
+}

flink-table/flink-sql-parser-hive/src/test/java/org/apache/flink/sql/parser/hive/FlinkHiveSqlParserImplTest.java

@@ -466,6 +466,11 @@ public void testAddJar() {
  sql("add JAR 'oss:https://path/helloworld.go'").ok("ADD JAR 'oss:https://path/helloworld.go'");
  }
 
+ @Test
+ public void testShowJars() {
+ sql("show jars").ok("SHOW JARS");
+ }
+
  @Test
  public void testLoadModule() {
  sql("load module hive").ok("LOAD MODULE `HIVE`");

flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd

@@ -71,6 +71,7 @@
  "org.apache.flink.sql.parser.dql.SqlShowDatabases"
  "org.apache.flink.sql.parser.dql.SqlShowCurrentDatabase"
  "org.apache.flink.sql.parser.dql.SqlShowFunctions"
+ "org.apache.flink.sql.parser.dql.SqlShowJars"
  "org.apache.flink.sql.parser.dql.SqlShowModules"
  "org.apache.flink.sql.parser.dql.SqlShowTables"
  "org.apache.flink.sql.parser.dql.SqlShowCreateTable"
@@ -105,6 +106,7 @@
  "FUNCTIONS"
  "IF"
  "JAR"
+ "JARS"
  "LOAD"
  "METADATA"
  "MODULES"
@@ -241,6 +243,7 @@
  "ISOLATION"
  "ISOYEAR"
  "JAR"
+ "JARS"
  "JAVA"
  "JSON"
  "K"
@@ -494,6 +497,7 @@
  "SqlUseModules()"
  "SqlRichExplain()"
  "SqlAddJar()"
+ "SqlShowJars()"
  "SqlSet()"
  "SqlReset()"
  ]

flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl

@@ -1665,6 +1665,20 @@ SqlAddJar SqlAddJar() :
  }
 }
 
+/**
+* Parses a show jars statement.
+* SHOW JARS;
+*/
+SqlShowJars SqlShowJars() :
+{
+}
+{
+ <SHOW> <JARS>
+ {
+ return new SqlShowJars(getPos());
+ }
+}
+
 /*
 * Parses a SET statement:
 * SET ['key' = 'value'];

flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/dql/SqlShowJars.java

@@ -0,0 +1,56 @@
+/*
+ * 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.sql.parser.dql;
+
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+
+import java.util.Collections;
+import java.util.List;
+
+/** SHOW JARS sql call. */
+public class SqlShowJars extends SqlCall {
+
+ public static final SqlSpecialOperator OPERATOR =
+ new SqlSpecialOperator("SHOW JARS", SqlKind.OTHER);
+
+ public SqlShowJars(SqlParserPos pos) {
+ super(pos);
+ }
+
+ @Override
+ public SqlOperator getOperator() {
+ return OPERATOR;
+ }
+
+ @Override
+ public List<SqlNode> getOperandList() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
+ writer.keyword("SHOW JARS");
+ }
+}

flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java

@@ -1343,6 +1343,11 @@ public void testAddJar() {
  sql("add JAR 'oss:https://path/helloworld.go'").ok("ADD JAR 'oss:https://path/helloworld.go'");
  }
 
+ @Test
+ public void testShowJars() {
+ sql("show jars").ok("SHOW JARS");
+ }
+
  @Test
  public void testSetReset() {
  sql("SET").ok("SET");

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

@@ -0,0 +1,30 @@
+/*
+ * 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.command;
+
+import org.apache.flink.table.operations.ShowOperation;
+
+/** Operation to describe a SHOW JARS statement. */
+public class ShowJarsOperation implements ShowOperation {
+
+ @Override
+ public String asSummaryString() {
+ return "SHOW JARS";
+ }
+}