[FLINK-29045][hive] Optimize error message in Flink SQL Client and Ga…

…teway when try to use Hive Dialect This closes apache#20695
huangxiaofeng10047 · Nov 3, 2022 · 2faf3d9 · 2faf3d9
1 parent f97ac45
commit 2faf3d9
Show file tree

Hide file tree

Showing 2 changed files with 35 additions and 10 deletions.
diff --git a/...ql-client/src/main/java/org/apache/flink/table/client/gateway/context/SessionContext.java b/...ql-client/src/main/java/org/apache/flink/table/client/gateway/context/SessionContext.java
@@ -24,6 +24,8 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.SqlDialect;
+import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.catalog.Catalog;
 import org.apache.flink.table.catalog.CatalogManager;
 import org.apache.flink.table.catalog.FunctionCatalog;
@@ -34,6 +36,7 @@
 import org.apache.flink.table.client.util.ClientClassloaderUtil;
 import org.apache.flink.table.client.util.ClientWrapperClassLoader;
 import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.TemporaryClassLoaderContext;
 
 import org.slf4j.Logger;
@@ -156,6 +159,14 @@ public void set(String key, String value) {
         } catch (Exception e) {
             // get error and reset the key with old value
             resetSessionConfigurationToDefault(originConfiguration);
+            if (value.equalsIgnoreCase(SqlDialect.HIVE.name())
+                    && e instanceof ValidationException) {
+                String additionErrorMsg =
+                        "Note: if you want to use Hive dialect, "
+                                + "please first move the jar `flink-table-planner_2.12` located in `FLINK_HOME/opt` "
+                                + "to `FLINK_HOME/lib` and then move out the jar `flink-table-planner-loader` from `FLINK_HOME/lib`.";
+                ExceptionUtils.updateDetailMessage(e, t -> t.getMessage() + additionErrorMsg);
+            }
             throw new SqlExecutionException(
                     String.format("Failed to set key %s with value %s.", key, value), e);
         }

diff --git a/...-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java b/...-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java
@@ -23,8 +23,10 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.SqlDialect;
 import org.apache.flink.table.api.TableConfig;
 import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
 import org.apache.flink.table.api.config.TableConfigOptions;
 import org.apache.flink.table.api.internal.TableEnvironmentInternal;
@@ -46,6 +48,7 @@
 import org.apache.flink.table.gateway.service.utils.SqlExecutionException;
 import org.apache.flink.table.module.ModuleManager;
 import org.apache.flink.table.resource.ResourceManager;
+import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.FlinkUserCodeClassLoaders;
 import org.apache.flink.util.MutableURLClassLoader;
 
@@ -293,16 +296,27 @@ private TableEnvironmentInternal createStreamTableEnvironment(
                         catalogManager,
                         functionCatalog);
 
-        return new StreamTableEnvironmentImpl(
-                catalogManager,
-                moduleManager,
-                resourceManager,
-                functionCatalog,
-                tableConfig,
-                env,
-                planner,
-                executor,
-                settings.isStreamingMode());
+        try {
+            return new StreamTableEnvironmentImpl(
+                    catalogManager,
+                    moduleManager,
+                    resourceManager,
+                    functionCatalog,
+                    tableConfig,
+                    env,
+                    planner,
+                    executor,
+                    settings.isStreamingMode());
+        } catch (ValidationException e) {
+            if (tableConfig.getSqlDialect() == SqlDialect.HIVE) {
+                String additionErrorMsg =
+                        "Note: if you want to use Hive dialect, "
+                                + "please first move the jar `flink-table-planner_2.12` located in `FLINK_HOME/opt` "
+                                + "to `FLINK_HOME/lib` and then move out the jar `flink-table-planner-loader` from `FLINK_HOME/lib`.";
+                ExceptionUtils.updateDetailMessage(e, t -> t.getMessage() + additionErrorMsg);
+            }
+            throw e;
+        }
     }
 
     private static Executor lookupExecutor(