This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new a6a7063cfaf [FLINK-28096][hive] Hive dialect support set variables
(#20000)
a6a7063cfaf is described below
commit a6a7063cfafd8cd5c1b04169839b92373bf96fde
Author: yuxia Luo <[email protected]>
AuthorDate: Mon Aug 1 14:12:32 2022 +0800
[FLINK-28096][hive] Hive dialect support set variables (#20000)
---
.../hive/HiveInternalOptions.java} | 27 +-
.../flink/table/operations/HiveSetOperation.java | 80 ++++++
.../delegation/hive/HiveDialectFactory.java | 2 +-
.../delegation/hive/HiveOperationExecutor.java | 90 +++++++
.../table/planner/delegation/hive/HiveParser.java | 83 ++++++
.../delegation/hive/copy/HiveSetProcessor.java | 284 +++++++++++++++++++++
.../flink/connectors/hive/HiveDialectITCase.java | 70 ++++-
7 files changed, 620 insertions(+), 16 deletions(-)
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveOperationExecutor.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveInternalOptions.java
similarity index 56%
copy from
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveOperationExecutor.java
copy to
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveInternalOptions.java
index 2eb02ba27a6..655f25b114b 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveOperationExecutor.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveInternalOptions.java
@@ -16,21 +16,20 @@
* limitations under the License.
*/
-package org.apache.flink.table.planner.delegation.hive;
+package org.apache.flink.connectors.hive;
-import org.apache.flink.table.api.internal.TableResultInternal;
-import org.apache.flink.table.delegation.ExtendedOperationExecutor;
-import org.apache.flink.table.operations.Operation;
+import org.apache.flink.configuration.ConfigOption;
-import java.util.Optional;
+import java.util.HashMap;
+import java.util.Map;
-/**
- * A Hive's operation executor used to execute operation in custom way instead
of Flink's
- * implementation.
- */
-public class HiveOperationExecutor implements ExtendedOperationExecutor {
- @Override
- public Optional<TableResultInternal> executeOperation(Operation operation)
{
- return Optional.empty();
- }
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/** This class holds internal configuration constants used by Hive connector
module. */
+public class HiveInternalOptions {
+ public static final ConfigOption<Map<String, String>> HIVE_VARIABLES =
+ key("__hive.variables__")
+ .mapType()
+ .defaultValue(new HashMap<>())
+ .withDescription("The config used to save the hive
variables set by Flink.");
}
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/operations/HiveSetOperation.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/operations/HiveSetOperation.java
new file mode 100644
index 00000000000..d675ba8df37
--- /dev/null
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/operations/HiveSetOperation.java
@@ -0,0 +1,80 @@
+/*
+ * 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://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;
+
+import javax.annotation.Nullable;
+
+import java.util.Optional;
+
+/** Hive's set operation. */
+public class HiveSetOperation implements Operation {
+ private final @Nullable String key;
+ private final @Nullable String value;
+ // for Hive's command "set -v"
+ private final boolean isVerbose;
+
+ public HiveSetOperation() {
+ this(null, null, false);
+ }
+
+ public HiveSetOperation(boolean isVerbose) {
+ this(null, null, isVerbose);
+ }
+
+ public HiveSetOperation(String key) {
+ this(key, null);
+ }
+
+ public HiveSetOperation(String key, String value) {
+ this(key, value, false);
+ }
+
+ public HiveSetOperation(String key, String value, boolean isVerbose) {
+ this.key = key;
+ this.value = value;
+ this.isVerbose = isVerbose;
+ }
+
+ public boolean isVerbose() {
+ return isVerbose;
+ }
+
+ public Optional<String> getKey() {
+ return Optional.ofNullable(key);
+ }
+
+ public Optional<String> getValue() {
+ return Optional.ofNullable(value);
+ }
+
+ @Override
+ public String asSummaryString() {
+ if (isVerbose) {
+ return "set -v";
+ }
+ StringBuilder sb = new StringBuilder("set");
+ if (key != null) {
+ sb.append(" ").append(key).append("=");
+ }
+ if (value != null) {
+ sb.append(value);
+ }
+ return sb.toString();
+ }
+}
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveDialectFactory.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveDialectFactory.java
index 10b8dce8f9a..08c2bc15e75 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveDialectFactory.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveDialectFactory.java
@@ -56,6 +56,6 @@ public class HiveDialectFactory implements DialectFactory {
@Override
public ExtendedOperationExecutor createExtendedOperationExecutor(Context
context) {
- return new HiveOperationExecutor();
+ return new HiveOperationExecutor(context.getCatalogManager(),
context.getPlannerContext());
}
}
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveOperationExecutor.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveOperationExecutor.java
index 2eb02ba27a6..459a1bbd7dd 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveOperationExecutor.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveOperationExecutor.java
@@ -18,19 +18,109 @@
package org.apache.flink.table.planner.delegation.hive;
+import org.apache.flink.connectors.hive.FlinkHiveException;
+import org.apache.flink.connectors.hive.HiveInternalOptions;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ResultKind;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.internal.TableResultImpl;
import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.flink.table.delegation.ExtendedOperationExecutor;
+import org.apache.flink.table.operations.HiveSetOperation;
import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.planner.delegation.PlannerContext;
+import org.apache.flink.table.planner.delegation.hive.copy.HiveSetProcessor;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.hive.conf.HiveConf;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
import java.util.Optional;
+import java.util.stream.Collectors;
/**
* A Hive's operation executor used to execute operation in custom way instead
of Flink's
* implementation.
*/
public class HiveOperationExecutor implements ExtendedOperationExecutor {
+
+ private final CatalogManager catalogManager;
+ private final Map<String, String> hiveVariables;
+ private final TableConfig tableConfig;
+
+ public HiveOperationExecutor(CatalogManager catalogManager, PlannerContext
plannerContext) {
+ this.catalogManager = catalogManager;
+ tableConfig = plannerContext.getFlinkContext().getTableConfig();
+ this.hiveVariables =
tableConfig.get(HiveInternalOptions.HIVE_VARIABLES);
+ }
+
@Override
public Optional<TableResultInternal> executeOperation(Operation operation)
{
+ if (operation instanceof HiveSetOperation) {
+ return executeHiveSetOperation((HiveSetOperation) operation);
+ }
return Optional.empty();
}
+
+ private Optional<TableResultInternal> executeHiveSetOperation(
+ HiveSetOperation hiveSetOperation) {
+ Catalog currentCatalog =
+
catalogManager.getCatalog(catalogManager.getCurrentCatalog()).orElse(null);
+ if (!(currentCatalog instanceof HiveCatalog)) {
+ throw new FlinkHiveException(
+ "Only support SET command when the current catalog is
HiveCatalog ing Hive dialect.");
+ }
+
+ HiveConf hiveConf = ((HiveCatalog) currentCatalog).getHiveConf();
+
+ if (!hiveSetOperation.getKey().isPresent() &&
!hiveSetOperation.getValue().isPresent()) {
+ List<String> options;
+ if (hiveSetOperation.isVerbose()) {
+ // set -v
+ options =
+ HiveSetProcessor.dumpOptions(
+ hiveConf.getAllProperties(), hiveConf,
hiveVariables, tableConfig);
+ } else {
+ // set
+ options =
+ HiveSetProcessor.dumpOptions(
+ hiveConf.getChangedProperties(),
+ hiveConf,
+ hiveVariables,
+ tableConfig);
+ }
+ return Optional.of(buildResultForShowVariable(options));
+ } else {
+ if (!hiveSetOperation.getValue().isPresent()) {
+ // set key
+ String option =
+ HiveSetProcessor.getVariable(
+ hiveConf, hiveVariables,
hiveSetOperation.getKey().get());
+ return
Optional.of(buildResultForShowVariable(Collections.singletonList(option)));
+ } else {
+ HiveSetProcessor.setVariable(
+ hiveConf,
+ hiveVariables,
+ hiveSetOperation.getKey().get(),
+ hiveSetOperation.getValue().get());
+ return Optional.of(TableResultImpl.TABLE_RESULT_OK);
+ }
+ }
+ }
+
+ private TableResultInternal buildResultForShowVariable(List<String>
results) {
+ List<Row> rows =
results.stream().map(Row::of).collect(Collectors.toList());
+ return TableResultImpl.builder()
+ .resultKind(ResultKind.SUCCESS)
+ .schema(ResolvedSchema.of(Column.physical("variables",
DataTypes.STRING())))
+ .data(rows)
+ .build();
+ }
}
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParser.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParser.java
index 6f8cc71b99c..79ffb6bb760 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParser.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParser.java
@@ -19,7 +19,9 @@
package org.apache.flink.table.planner.delegation.hive;
import org.apache.flink.connectors.hive.FlinkHiveException;
+import org.apache.flink.connectors.hive.HiveInternalOptions;
import org.apache.flink.table.api.SqlParserException;
+import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CatalogManager;
@@ -29,6 +31,7 @@ import
org.apache.flink.table.catalog.hive.client.HiveShimLoader;
import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
import org.apache.flink.table.module.hive.udf.generic.HiveGenericUDFGrouping;
import org.apache.flink.table.operations.ExplainOperation;
+import org.apache.flink.table.operations.HiveSetOperation;
import org.apache.flink.table.operations.ModifyOperation;
import org.apache.flink.table.operations.NopOperation;
import org.apache.flink.table.operations.Operation;
@@ -57,6 +60,7 @@ import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.lockmgr.LockException;
import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.processors.HiveCommand;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -72,9 +76,13 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.function.Supplier;
+import static
org.apache.flink.table.planner.delegation.hive.copy.HiveSetProcessor.startWithHiveSpecialVariablePrefix;
+
/** A Parser that uses Hive's planner to parse a statement. */
public class HiveParser extends ParserImpl {
@@ -163,6 +171,8 @@ public class HiveParser extends ParserImpl {
private final FrameworkConfig frameworkConfig;
private final SqlFunctionConverter funcConverter;
private final HiveParserDMLHelper dmlHelper;
+ private final TableConfig tableConfig;
+ private final Map<String, String> hiveVariables;
HiveParser(
CatalogManager catalogManager,
@@ -183,6 +193,8 @@ public class HiveParser extends ParserImpl {
frameworkConfig.getOperatorTable(),
catalogReader.nameMatcher());
this.dmlHelper = new HiveParserDMLHelper(plannerContext,
funcConverter, catalogManager);
+ this.tableConfig = plannerContext.getFlinkContext().getTableConfig();
+ this.hiveVariables =
tableConfig.get(HiveInternalOptions.HIVE_VARIABLES);
}
@Override
@@ -194,6 +206,11 @@ public class HiveParser extends ParserImpl {
LOG.warn("Current catalog is not HiveCatalog. Falling back to
Flink's planner.");
return super.parse(statement);
}
+
+ Optional<Operation> nonSqlOperation =
tryProcessHiveNonSqlStatement(statement);
+ if (nonSqlOperation.isPresent()) {
+ return Collections.singletonList(nonSqlOperation.get());
+ }
HiveConf hiveConf = new HiveConf(((HiveCatalog)
currentCatalog).getHiveConf());
hiveConf.setVar(HiveConf.ConfVars.DYNAMICPARTITIONINGMODE,
"nonstrict");
hiveConf.set("hive.allow.udf.load.on.demand", "false");
@@ -211,6 +228,72 @@ public class HiveParser extends ParserImpl {
}
}
+ private Optional<Operation> tryProcessHiveNonSqlStatement(String
statement) {
+ String[] commandTokens = statement.split("\\s+");
+ HiveCommand hiveCommand = HiveCommand.find(commandTokens);
+ if (hiveCommand != null) {
+ if (hiveCommand == HiveCommand.SET) {
+ return Optional.of(
+ processSetCmd(
+ statement,
statement.substring(commandTokens[0].length()).trim()));
+ } else if (hiveCommand == HiveCommand.RESET) {
+ return Optional.of(super.parse(statement).get(0));
+ } else {
+ throw new UnsupportedOperationException(
+ String.format("The Hive command %s is not supported.",
hiveCommand));
+ }
+ }
+ return Optional.empty();
+ }
+
+ private Operation processSetCmd(String originCmd, String setCmdArgs) {
+ String nwcmd = setCmdArgs.trim();
+ // the set command may end with ";" since it won't be removed by Flink
SQL CLI,
+ // so, we need to remove ";"
+ if (nwcmd.endsWith(";")) {
+ nwcmd = nwcmd.substring(0, nwcmd.length() - 1);
+ }
+
+ if (nwcmd.equals("")) {
+ return new HiveSetOperation();
+ }
+ if (nwcmd.equals("-v")) {
+ return new HiveSetOperation(true);
+ }
+
+ String[] part = new String[2];
+ int eqIndex = nwcmd.indexOf('=');
+ if (nwcmd.contains("=")) {
+ if (eqIndex == nwcmd.length() - 1) { // x=
+ part[0] = nwcmd.substring(0, nwcmd.length() - 1);
+ part[1] = "";
+ } else { // x=y
+ part[0] = nwcmd.substring(0, eqIndex).trim();
+ part[1] = nwcmd.substring(eqIndex + 1).trim();
+ if (!startWithHiveSpecialVariablePrefix(part[0])) {
+ // TODO:
+ // currently, for the command set key=value, we will fall
to
+ // Flink's implementation, otherwise, user will have no
way to switch dialect.
+ // need to figure out whether we should also set the value
in HiveConf which is
+ // Hive's implementation
+ LOG.warn(
+ "The command 'set {}={}' will only set Flink's
table config,"
+ + " and if you want to set the variable to
Hive's conf, please use the command like 'set hiveconf:{}={}'.",
+ part[0],
+ part[1],
+ part[0],
+ part[1]);
+ return super.parse(originCmd).get(0);
+ }
+ }
+ if (part[0].equals("silent")) {
+ throw new UnsupportedOperationException("Unsupported command
'set silent'.");
+ }
+ return new HiveSetOperation(part[0], part[1]);
+ }
+ return new HiveSetOperation(nwcmd);
+ }
+
private List<Operation> processCmd(
String cmd, HiveConf hiveConf, HiveShim hiveShim, HiveCatalog
hiveCatalog) {
try {
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveSetProcessor.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveSetProcessor.java
new file mode 100644
index 00000000000..51ad1dddd17
--- /dev/null
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveSetProcessor.java
@@ -0,0 +1,284 @@
+/*
+ * 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://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.planner.delegation.hive.copy;
+
+import org.apache.flink.connectors.hive.FlinkHiveException;
+import org.apache.flink.table.api.TableConfig;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.VariableSubstitution;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.SortedMap;
+import java.util.TreeMap;
+
+import static org.apache.hadoop.hive.conf.SystemVariables.ENV_PREFIX;
+import static org.apache.hadoop.hive.conf.SystemVariables.HIVECONF_PREFIX;
+import static org.apache.hadoop.hive.conf.SystemVariables.HIVEVAR_PREFIX;
+import static org.apache.hadoop.hive.conf.SystemVariables.METACONF_PREFIX;
+import static org.apache.hadoop.hive.conf.SystemVariables.SYSTEM_PREFIX;
+
+/** Counterpart of hive's {@link
org.apache.hadoop.hive.ql.processors.SetProcessor}. */
+public class HiveSetProcessor {
+
+ private static final String[] PASSWORD_STRINGS = new String[] {"password",
"paswd", "pswd"};
+ private static final String FLINK_PREFIX = "flink:";
+
+ /** Set variable following Hive's implementation. */
+ public static void setVariable(
+ HiveConf hiveConf, Map<String, String> hiveVariables, String
varname, String varvalue) {
+ if (varname.startsWith(ENV_PREFIX)) {
+ throw new UnsupportedOperationException("env:* variables can not
be set.");
+ } else if (varname.startsWith(SYSTEM_PREFIX)) {
+ String propName = varname.substring(SYSTEM_PREFIX.length());
+ System.getProperties()
+ .setProperty(
+ propName,
+ new VariableSubstitution(() -> hiveVariables)
+ .substitute(hiveConf, varvalue));
+ } else if (varname.startsWith(HIVECONF_PREFIX)) {
+ String propName = varname.substring(HIVECONF_PREFIX.length());
+ setConf(hiveConf, hiveVariables, varname, propName, varvalue);
+ } else if (varname.startsWith(HIVEVAR_PREFIX)) {
+ String propName = varname.substring(HIVEVAR_PREFIX.length());
+ hiveVariables.put(
+ propName,
+ new VariableSubstitution(() ->
hiveVariables).substitute(hiveConf, varvalue));
+ } else if (varname.startsWith(METACONF_PREFIX)) {
+ String propName = varname.substring(METACONF_PREFIX.length());
+ try {
+ Hive hive = Hive.get(hiveConf);
+ hive.setMetaConf(
+ propName,
+ new VariableSubstitution(() -> hiveVariables)
+ .substitute(hiveConf, varvalue));
+ } catch (HiveException e) {
+ throw new FlinkHiveException(
+ String.format("'SET %s=%s' FAILED.", varname,
varvalue), e);
+ }
+ } else {
+ setConf(hiveConf, hiveVariables, varname, varname, varvalue);
+ }
+ }
+
+ /**
+ * check whether the variable's name is started with the special variable
prefix that Hive
+ * reserves.
+ */
+ public static boolean startWithHiveSpecialVariablePrefix(String varname) {
+ String[] hiveSpecialVariablePrefix =
+ new String[] {
+ ENV_PREFIX, SYSTEM_PREFIX, HIVECONF_PREFIX,
HIVEVAR_PREFIX, METACONF_PREFIX
+ };
+ for (String prefix : hiveSpecialVariablePrefix) {
+ if (varname.startsWith(prefix)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static void setConf(
+ HiveConf hiveConf,
+ Map<String, String> hiveVariables,
+ String varname,
+ String key,
+ String varvalue) {
+ String value = new VariableSubstitution(() ->
hiveVariables).substitute(hiveConf, varvalue);
+ if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVECONFVALIDATION)) {
+ HiveConf.ConfVars confVars = HiveConf.getConfVars(key);
+ if (confVars != null) {
+ if (!confVars.isType(value)) {
+ String message =
+ String.format(
+ "'SET %s=%s' FAILED because %s expects %s
type value.",
+ varname, varvalue, key,
confVars.typeString());
+ throw new IllegalArgumentException(message);
+ }
+ String fail = confVars.validate(value);
+ if (fail != null) {
+ String message =
+ String.format(
+ "'SET %s=%s' FAILED in validation : %s.",
+ varname, varvalue, fail);
+ throw new IllegalArgumentException(message);
+ }
+ }
+ }
+ hiveConf.verifyAndSet(key, value);
+ }
+
+ public static String getVariable(
+ HiveConf hiveConf, Map<String, String> hiveVariables, String
varname) {
+ if (varname.equals("silent")) {
+ return "silent is not a valid variable";
+ }
+ if (varname.startsWith(SYSTEM_PREFIX)) {
+ String propName = varname.substring(SYSTEM_PREFIX.length());
+ String result = System.getProperty(propName);
+ if (result != null) {
+ if (isHidden(propName)) {
+ return SYSTEM_PREFIX + propName + " is a hidden config";
+ } else {
+ return SYSTEM_PREFIX + propName + "=" + result;
+ }
+ } else {
+ return propName + " is undefined as a system property";
+ }
+ } else if (varname.indexOf(ENV_PREFIX) == 0) {
+ String var = varname.substring(ENV_PREFIX.length());
+ if (System.getenv(var) != null) {
+ if (isHidden(var)) {
+ return ENV_PREFIX + var + " is a hidden config";
+ } else {
+ return ENV_PREFIX + var + "=" + System.getenv(var);
+ }
+ } else {
+ return varname + " is undefined as an environmental variable";
+ }
+ } else if (varname.indexOf(HIVECONF_PREFIX) == 0) {
+ String var = varname.substring(HIVECONF_PREFIX.length());
+ if (hiveConf.isHiddenConfig(var)) {
+ return HIVECONF_PREFIX + var + " is a hidden config";
+ }
+ if (hiveConf.get(var) != null) {
+ return HIVECONF_PREFIX + var + "=" + hiveConf.get(var);
+ } else {
+ return varname + " is undefined as a hive configuration
variable";
+ }
+ } else if (varname.indexOf(HIVEVAR_PREFIX) == 0) {
+ String var = varname.substring(HIVEVAR_PREFIX.length());
+ if (hiveVariables.get(var) != null) {
+ return HIVEVAR_PREFIX + var + "=" + hiveVariables.get(var);
+ } else {
+ return varname + " is undefined as a hive variable";
+ }
+ } else if (varname.indexOf(METACONF_PREFIX) == 0) {
+ String var = varname.substring(METACONF_PREFIX.length());
+ String value;
+ try {
+ Hive hive = Hive.get(hiveConf);
+ value = hive.getMetaConf(var);
+ } catch (HiveException e) {
+ throw new FlinkHiveException(
+ String.format("Failed to get variable for %s.",
varname), e);
+ }
+
+ if (value != null) {
+ return METACONF_PREFIX + var + "=" + value;
+ } else {
+ return varname + " is undefined as a hive meta variable";
+ }
+ } else {
+ return dumpOption(hiveConf, hiveVariables, varname);
+ }
+ }
+
+ /*
+ * Checks if the value contains any of the PASSWORD_STRINGS and if yes
+ * return true
+ */
+ private static boolean isHidden(String key) {
+ for (String p : PASSWORD_STRINGS) {
+ if (key.toLowerCase().contains(p)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static String dumpOption(
+ HiveConf hiveConf, Map<String, String> hiveVariables, String s) {
+ if (hiveConf.isHiddenConfig(s)) {
+ return s + " is a hidden config";
+ } else if (hiveConf.get(s) != null) {
+ return s + "=" + hiveConf.get(s);
+ } else if (hiveVariables.containsKey(s)) {
+ return s + "=" + hiveVariables.get(s);
+ } else {
+ return s + " is undefined";
+ }
+ }
+
+ public static List<String> dumpOptions(
+ Properties p,
+ HiveConf hiveConf,
+ Map<String, String> hiveVariables,
+ TableConfig tableConfig) {
+ SortedMap<String, String> sortedMap = new TreeMap<>();
+ List<String> optionsList = new ArrayList<>();
+ for (Object one : p.keySet()) {
+ String oneProp = (String) one;
+ String oneValue = p.getProperty(oneProp);
+ if (hiveConf.isHiddenConfig(oneProp)) {
+ continue;
+ }
+ sortedMap.put(oneProp, oneValue);
+ }
+
+ // Inserting hive variables
+ for (String s : hiveVariables.keySet()) {
+ sortedMap.put(HIVEVAR_PREFIX + s, hiveVariables.get(s));
+ }
+
+ for (Map.Entry<String, String> entries : sortedMap.entrySet()) {
+ optionsList.add(entries.getKey() + "=" + entries.getValue());
+ }
+
+ for (Map.Entry<String, String> entry :
mapToSortedMap(System.getenv()).entrySet()) {
+ if (isHidden(entry.getKey())) {
+ continue;
+ }
+ optionsList.add(ENV_PREFIX + entry.getKey() + "=" +
entry.getValue());
+ }
+
+ for (Map.Entry<String, String> entry :
+ propertiesToSortedMap(System.getProperties()).entrySet()) {
+ if (isHidden(entry.getKey())) {
+ continue;
+ }
+ optionsList.add(SYSTEM_PREFIX + entry.getKey() + "=" +
entry.getValue());
+ }
+
+ // Insert Flink table config variable
+ for (Map.Entry<String, String> entry :
+
mapToSortedMap(tableConfig.getConfiguration().toMap()).entrySet()) {
+ optionsList.add(FLINK_PREFIX + entry.getKey() + "=" +
entry.getValue());
+ }
+
+ return optionsList;
+ }
+
+ private static SortedMap<String, String> mapToSortedMap(Map<String,
String> data) {
+ return new TreeMap<>(data);
+ }
+
+ private static SortedMap<String, String> propertiesToSortedMap(Properties
p) {
+ SortedMap<String, String> sortedPropMap = new TreeMap<>();
+ for (Map.Entry<Object, Object> entry : p.entrySet()) {
+ sortedPropMap.put((String) entry.getKey(), (String)
entry.getValue());
+ }
+ return sortedPropMap;
+ }
+}
diff --git
a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java
b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java
index 618f3d69044..baf8c2ab4a5 100644
---
a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java
+++
b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java
@@ -40,6 +40,7 @@ import org.apache.flink.table.delegation.Parser;
import org.apache.flink.table.functions.hive.HiveGenericUDTFTest;
import
org.apache.flink.table.functions.hive.util.TestSplitUDTFInitializeWithStructObjectInspector;
import org.apache.flink.table.operations.DescribeTableOperation;
+import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.command.ClearOperation;
import org.apache.flink.table.operations.command.HelpOperation;
import org.apache.flink.table.operations.command.QuitOperation;
@@ -65,6 +66,7 @@ import org.apache.hadoop.hive.ql.io.RCFileOutputFormat;
import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat;
import org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat;
import org.apache.hadoop.hive.ql.io.orc.OrcSerde;
+import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFCount;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFAbs;
@@ -159,7 +161,6 @@ public class HiveDialectITCase {
assertThat(parser).isInstanceOf(HiveParser.class);
assertThat(parser.parse("HELP").get(0)).isInstanceOf(HelpOperation.class);
assertThat(parser.parse("clear").get(0)).isInstanceOf(ClearOperation.class);
-
assertThat(parser.parse("SET").get(0)).isInstanceOf(SetOperation.class);
assertThat(parser.parse("ResET").get(0)).isInstanceOf(ResetOperation.class);
assertThat(parser.parse("Exit").get(0)).isInstanceOf(QuitOperation.class);
}
@@ -1018,6 +1019,73 @@ public class HiveDialectITCase {
badMacroName);
}
+ @Test
+ public void testSetCommand() throws Exception {
+ // test set system:
+ tableEnv.executeSql("set system:xxx=5");
+ assertThat(System.getProperty("xxx")).isEqualTo("5");
+
+ // test set hiveconf:
+ tableEnv.executeSql("set hiveconf:yyy=${system:xxx}");
+ assertThat(hiveCatalog.getHiveConf().get("yyy")).isEqualTo("5");
+
+ // test set hivevar:
+ tableEnv.executeSql("set hivevar:a=1");
+ tableEnv.executeSql("set hiveconf:zzz=${hivevar:a}");
+ assertThat(hiveCatalog.getHiveConf().get("zzz")).isEqualTo("1");
+
+ // test the hivevar still exists when we renew the sql parser
+ tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
+ tableEnv.executeSql("show tables");
+ tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
+ tableEnv.executeSql("set hiveconf:zzz1=${hivevar:a}");
+ assertThat(hiveCatalog.getHiveConf().get("zzz1")).isEqualTo("1");
+
+ // test set metaconf:
+ tableEnv.executeSql("set
metaconf:hive.metastore.try.direct.sql=false");
+ Hive hive = Hive.get(hiveCatalog.getHiveConf());
+
assertThat(hive.getMetaConf("hive.metastore.try.direct.sql")).isEqualTo("false");
+
+ // test 'set xxx = xxx' should be pared as Flink SetOperation
+ TableEnvironmentInternal tableEnvInternal = (TableEnvironmentInternal)
tableEnv;
+ Parser parser = tableEnvInternal.getParser();
+ Operation operation = parser.parse("set
user.name=hive_test_user").get(0);
+ assertThat(operation).isInstanceOf(SetOperation.class);
+ assertThat(operation.asSummaryString()).isEqualTo("SET
user.name=hive_test_user");
+
+ // test 'set xxx='
+ tableEnv.executeSql("set user.name=");
+ assertThat(hiveCatalog.getHiveConf().get("user.name")).isEmpty();
+
+ // test 'set xxx'
+ List<Row> rows =
+ CollectionUtil.iteratorToList(tableEnv.executeSql("set
system:xxx").collect());
+ assertThat(rows.toString()).isEqualTo("[+I[system:xxx=5]]");
+
+ // test 'set'
+ rows =
CollectionUtil.iteratorToList(tableEnv.executeSql("set").collect());
+ assertThat(rows.toString())
+ .contains("system:xxx=5")
+ .contains("env:PATH=" + System.getenv("PATH"))
+ .contains("flink:execution.runtime-mode=BATCH")
+ .contains("hivevar:a=1")
+ .contains("common-key=common-val");
+
+ // test 'set -v'
+ rows = CollectionUtil.iteratorToList(tableEnv.executeSql("set
-v").collect());
+ assertThat(rows.toString())
+ .contains("system:xxx=5")
+ .contains("env:PATH=" + System.getenv("PATH"))
+ .contains("flink:execution.runtime-mode=BATCH")
+ .contains("hivevar:a=1")
+ .contains("fs.defaultFS=file:///");
+
+ // test set env isn't supported
+ assertThatThrownBy(() -> tableEnv.executeSql("set env:xxx=yyy"))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("env:* variables can not be set.");
+ }
+
@Test
public void testUnsupportedOperation() {
List<String> statements =
