You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2020/11/24 02:56:37 UTC
[flink] branch master updated: [FLINK-19653][hive] Reduce our
dependency on hive runner for tests
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 e9b05e7 [FLINK-19653][hive] Reduce our dependency on hive runner for tests
e9b05e7 is described below
commit e9b05e723fb02d9a6dae607ef28244eca6e9edc8
Author: Rui Li <li...@apache.org>
AuthorDate: Tue Nov 24 10:53:43 2020 +0800
[FLINK-19653][hive] Reduce our dependency on hive runner for tests
This closes #14123
---
.../flink/connectors/hive/HiveTableSource.java | 13 -
...iveRunner.java => FlinkEmbeddedHiveRunner.java} | 193 +-----
...xt.java => FlinkEmbeddedHiveServerContext.java} | 32 +-
.../connectors/hive/FlinkStandaloneHiveRunner.java | 284 +--------
...eConnectorITCase.java => HiveRunnerITCase.java} | 663 +++++++++------------
.../flink/connectors/hive/HiveTableSinkITCase.java | 253 +-------
.../connectors/hive/HiveTableSourceITCase.java | 160 ++---
.../hive/TableEnvHiveConnectorITCase.java | 483 +--------------
.../table/catalog/hive/HiveCatalogITCase.java | 13 +-
.../catalog/hive/HiveCatalogUseBlinkITCase.java | 20 +-
.../flink/table/catalog/hive/HiveTestUtils.java | 97 +--
11 files changed, 463 insertions(+), 1748 deletions(-)
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
index f158625..6b87c90 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
@@ -27,7 +27,6 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.connectors.hive.read.HiveContinuousPartitionFetcher;
import org.apache.flink.connectors.hive.read.HivePartitionFetcherContextBase;
-import org.apache.flink.connectors.hive.read.HiveTableInputFormat;
import org.apache.flink.connectors.hive.util.HivePartitionUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
@@ -215,18 +214,6 @@ public class HiveTableSource implements
STREAMING_SOURCE_ENABLE.defaultValue().toString()));
}
- @VisibleForTesting
- HiveTableInputFormat getInputFormat(List<HiveTablePartition> allHivePartitions, boolean useMapRedReader) {
- return new HiveTableInputFormat(
- jobConf,
- catalogTable,
- allHivePartitions,
- projectedFields,
- limit,
- hiveVersion,
- useMapRedReader);
- }
-
protected TableSchema getTableSchema() {
return catalogTable.getSchema();
}
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkStandaloneHiveRunner.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkEmbeddedHiveRunner.java
similarity index 62%
copy from flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkStandaloneHiveRunner.java
copy to flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkEmbeddedHiveRunner.java
index 5caab72..0d528f2 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkStandaloneHiveRunner.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkEmbeddedHiveRunner.java
@@ -18,11 +18,8 @@
package org.apache.flink.connectors.hive;
-import org.apache.flink.table.catalog.hive.HiveTestUtils;
import org.apache.flink.util.Preconditions;
-import org.apache.flink.shaded.guava18.com.google.common.base.Joiner;
-import org.apache.flink.shaded.guava18.com.google.common.base.Throwables;
import org.apache.flink.shaded.guava18.com.google.common.io.Resources;
import com.klarna.hiverunner.HiveServerContainer;
@@ -37,8 +34,6 @@ import com.klarna.hiverunner.annotations.HiveSetupScript;
import com.klarna.hiverunner.builder.HiveShellBuilder;
import com.klarna.hiverunner.config.HiveRunnerConfig;
import com.klarna.reflection.ReflectionUtils;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.HiveMetaStore;
import org.junit.Ignore;
import org.junit.internal.AssumptionViolatedException;
import org.junit.internal.runners.model.EachTestNotifier;
@@ -54,79 +49,43 @@ import org.junit.runners.model.Statement;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
import java.lang.reflect.Field;
-import java.net.ConnectException;
-import java.net.InetSocketAddress;
-import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
-import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.Future;
-import java.util.concurrent.FutureTask;
-import java.util.concurrent.TimeUnit;
-
-import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVEHISTORYFILELOC;
-import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_IN_TEST;
-import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.LOCALSCRATCHDIR;
-import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTORECONNECTURLKEY;
-import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTOREWAREHOUSE;
-import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.SCRATCHDIR;
+
import static org.reflections.ReflectionUtils.withAnnotation;
/**
* JUnit 4 runner that runs hive sql on a HiveServer residing in this JVM. No external dependencies needed.
- * Inspired by StandaloneHiveRunner.java (almost copied), just using local meta store server instead of embedded
- * hive meta store.
+ * Inspired by StandaloneHiveRunner.java (almost copied).
*/
-public class FlinkStandaloneHiveRunner extends BlockJUnit4ClassRunner {
- private static final Logger LOGGER = LoggerFactory.getLogger(FlinkStandaloneHiveRunner.class);
- private static final Duration HMS_START_TIMEOUT = Duration.ofSeconds(90);
- private Future<Void> hmsWatcher;
- private int hmsPort;
+public class FlinkEmbeddedHiveRunner extends BlockJUnit4ClassRunner {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FlinkEmbeddedHiveRunner.class);
private HiveShellContainer container;
- private HiveRunnerConfig config = new HiveRunnerConfig();
+ private final HiveRunnerConfig config = new HiveRunnerConfig();
+ protected HiveServerContext context;
- public FlinkStandaloneHiveRunner(Class<?> clazz) throws InitializationError {
+ public FlinkEmbeddedHiveRunner(Class<?> clazz) throws InitializationError {
super(clazz);
}
@Override
protected List<TestRule> classRules() {
+ // need to load hive runner config before the context is inited
+ loadAnnotatesHiveRunnerConfig(getTestClass().getJavaClass());
final TemporaryFolder temporaryFolder = new TemporaryFolder();
- try {
- hmsPort = HiveTestUtils.getFreePort();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- HiveServerContext context = new FlinkStandaloneHiveServerContext(temporaryFolder, config, hmsPort);
+ context = new FlinkEmbeddedHiveServerContext(temporaryFolder, config);
List<TestRule> rules = super.classRules();
- ExternalResource hms = new ExternalResource() {
- @Override
- protected void before() throws Throwable {
- LOGGER.info("Setting up {} in {}", getName(), temporaryFolder.getRoot().getAbsolutePath());
- hmsWatcher = startHMS(getTestClass().getJavaClass(), context, hmsPort);
- }
-
- @Override
- protected void after() {
- if (hmsWatcher != null) {
- hmsWatcher.cancel(true);
- }
- }
- };
ExternalResource hiveShell = new ExternalResource() {
@Override
protected void before() throws Throwable {
@@ -139,7 +98,6 @@ public class FlinkStandaloneHiveRunner extends BlockJUnit4ClassRunner {
}
};
rules.add(hiveShell);
- rules.add(hms);
rules.add(temporaryFolder);
return rules;
}
@@ -193,6 +151,8 @@ public class FlinkStandaloneHiveRunner extends BlockJUnit4ClassRunner {
private HiveShellContainer createHiveServerContainer(final Class testClass, HiveServerContext context)
throws Exception {
+ context.init();
+
final HiveServerContainer hiveServerContainer = new HiveServerContainer(context);
HiveShellBuilder hiveShellBuilder = new HiveShellBuilder();
@@ -356,131 +316,4 @@ public class FlinkStandaloneHiveRunner extends BlockJUnit4ClassRunner {
boolean isAutoStart();
}
-
- /**
- * Launches HMS process and returns a Future representing that process.
- */
- private Future<Void> startHMS(Class testClass, HiveServerContext context, int port) throws Exception {
- // need to load hive runner config before the context is inited
- loadAnnotatesHiveRunnerConfig(testClass);
- context.init();
- HiveConf outsideConf = context.getHiveConf();
- List<String> args = new ArrayList<>();
- String javaHome = System.getProperty("java.home");
- args.add(Joiner.on(File.separator).join(javaHome, "bin", "java"));
- // set classpath
- args.add("-cp");
- args.add(System.getProperty("java.class.path"));
-
- // set sys properties
- args.add(hiveCmdLineConfig(METASTOREWAREHOUSE.varname, outsideConf.getVar(METASTOREWAREHOUSE)));
- args.add(hiveCmdLineConfig(SCRATCHDIR.varname, outsideConf.getVar(SCRATCHDIR)));
- args.add(hiveCmdLineConfig(LOCALSCRATCHDIR.varname, outsideConf.getVar(LOCALSCRATCHDIR)));
- args.add(hiveCmdLineConfig(HIVEHISTORYFILELOC.varname, outsideConf.getVar(HIVEHISTORYFILELOC)));
- // The following config is removed in Hive 3.1.0.
- args.add(hiveCmdLineConfig("hive.warehouse.subdir.inherit.perms",
- String.valueOf(outsideConf.getBoolean("hive.warehouse.subdir.inherit.perms", true))));
- args.add(hiveCmdLineConfig("hadoop.tmp.dir", outsideConf.get("hadoop.tmp.dir")));
- args.add(hiveCmdLineConfig("test.log.dir", outsideConf.get("test.log.dir")));
- String metaStorageUrl = "jdbc:derby:memory:" + UUID.randomUUID().toString();
- args.add(hiveCmdLineConfig(METASTORECONNECTURLKEY.varname, metaStorageUrl + ";create=true"));
- // config derby.log file
- File derbyLog = File.createTempFile("derby", ".log");
- derbyLog.deleteOnExit();
- args.add(hiveCmdLineConfig("derby.stream.error.file", derbyLog.getAbsolutePath()));
- // config whether in test
- if (outsideConf.getBoolVar(HIVE_IN_TEST)) {
- args.add(hiveCmdLineConfig(HIVE_IN_TEST.varname, "true"));
- }
-
- args.add(HiveMetaStore.class.getCanonicalName());
- args.add("-p");
- args.add(String.valueOf(port));
-
- ProcessBuilder builder = new ProcessBuilder(args);
- Process process = builder.start();
-
- try {
- Thread inLogger = new Thread(new LogRedirect(process.getInputStream(), LOGGER));
- Thread errLogger = new Thread(new LogRedirect(process.getErrorStream(), LOGGER));
- inLogger.setDaemon(true);
- inLogger.setName("HMS-IN-Logger");
- errLogger.setDaemon(true);
- errLogger.setName("HMS-ERR-Logger");
- inLogger.start();
- errLogger.start();
-
- FutureTask<Void> res = new FutureTask<>(() -> {
- try {
- int r = process.waitFor();
- inLogger.join();
- errLogger.join();
- if (r != 0) {
- throw new RuntimeException("HMS process exited with " + r);
- }
- } catch (InterruptedException e) {
- LOGGER.info("Shutting down HMS");
- } finally {
- if (process.isAlive()) {
- // give it a chance to terminate gracefully
- process.destroy();
- try {
- process.waitFor(5, TimeUnit.SECONDS);
- } catch (InterruptedException e) {
- LOGGER.info("Interrupted waiting for HMS to shut down, killing it forcibly");
- }
- process.destroyForcibly();
- }
- }
- }, null);
- Thread thread = new Thread(res);
- thread.setName("HMS-Watcher");
- // we need the watcher thread to kill HMS, don't make it daemon
- thread.setDaemon(false);
- thread.start();
- waitForHMSStart(port);
- return res;
- } catch (Throwable e) {
- // make sure to kill the process in case anything goes wrong
- process.destroyForcibly();
- throw e;
- }
- }
-
- private static void waitForHMSStart(int port) throws Exception {
- final long deadline = System.currentTimeMillis() + HMS_START_TIMEOUT.toMillis();
- while (System.currentTimeMillis() < deadline) {
- try (SocketChannel channel = SocketChannel.open(new InetSocketAddress("localhost", port))) {
- LOGGER.info("HMS started at port {}", port);
- return;
- } catch (ConnectException e) {
- LOGGER.info("Waiting for HMS to start...");
- Thread.sleep(1000);
- }
- }
- throw new java.util.concurrent.TimeoutException("Timeout waiting for HMS to start");
- }
-
- private static String hiveCmdLineConfig(String name, String val) {
- return String.format("-D%s=%s", name, val);
- }
-
- private static class LogRedirect implements Runnable {
- private final InputStream inputStream;
- private final Logger logger;
-
- LogRedirect(InputStream inputStream, Logger logger) {
- this.inputStream = inputStream;
- this.logger = logger;
- }
-
- @Override
- public void run() {
- try {
- new BufferedReader(new InputStreamReader(inputStream)).lines().forEach(logger::info);
- } catch (Exception e) {
- logger.error(Throwables.getStackTraceAsString(e));
- }
- }
- }
}
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkStandaloneHiveServerContext.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkEmbeddedHiveServerContext.java
similarity index 90%
rename from flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkStandaloneHiveServerContext.java
rename to flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkEmbeddedHiveServerContext.java
index 12d8368..58bf9b4 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkStandaloneHiveServerContext.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkEmbeddedHiveServerContext.java
@@ -18,19 +18,20 @@
package org.apache.flink.connectors.hive;
+import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
+
import com.klarna.hiverunner.HiveServerContext;
import com.klarna.hiverunner.config.HiveRunnerConfig;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hive.conf.HiveConf;
import org.junit.rules.TemporaryFolder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Map;
+import java.util.UUID;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HADOOPBIN;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVECONVERTJOIN;
@@ -45,6 +46,7 @@ import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_INFER_BUCKET_SO
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_SERVER2_LOGGING_OPERATION_ENABLED;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.LOCALSCRATCHDIR;
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTORECONNECTURLKEY;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTOREWAREHOUSE;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTORE_VALIDATE_COLUMNS;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTORE_VALIDATE_CONSTRAINTS;
@@ -52,27 +54,19 @@ import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTORE_VALIDATE_T
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.SCRATCHDIR;
/**
- * HiveServerContext used by FlinkStandaloneHiveRunner.
+ * HiveServerContext used by FlinkEmbeddedHiveRunner.
*/
-public class FlinkStandaloneHiveServerContext implements HiveServerContext {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(FlinkStandaloneHiveServerContext.class);
+public class FlinkEmbeddedHiveServerContext implements HiveServerContext {
- private HiveConf hiveConf = new HiveConf();
+ private final HiveConf hiveConf = new HiveConf();
private final TemporaryFolder basedir;
private final HiveRunnerConfig hiveRunnerConfig;
private boolean inited = false;
- private final int hmsPort;
- FlinkStandaloneHiveServerContext(TemporaryFolder basedir, HiveRunnerConfig hiveRunnerConfig, int hmsPort) {
+ FlinkEmbeddedHiveServerContext(TemporaryFolder basedir, HiveRunnerConfig hiveRunnerConfig) {
this.basedir = basedir;
this.hiveRunnerConfig = hiveRunnerConfig;
- this.hmsPort = hmsPort;
- }
-
- private String toHmsURI() {
- return "thrift://localhost:" + hmsPort;
}
@Override
@@ -179,7 +173,15 @@ public class FlinkStandaloneHiveServerContext implements HiveServerContext {
throw new RuntimeException(e);
}
- hiveConf.set("hive.metastore.uris", toHmsURI());
+ // No pooling needed. This will save us a lot of threads
+ // hive-2.1.1 doesn't allow 'none'...
+ if (!HiveShimLoader.getHiveVersion().equals("2.1.1")) {
+ hiveConf.set("datanucleus.connectionPoolingType", "None");
+ }
+
+ // set JDO configs
+ String jdoConnectionURL = "jdbc:derby:memory:" + UUID.randomUUID().toString();
+ hiveConf.setVar(METASTORECONNECTURLKEY, jdoConnectionURL + ";create=true");
hiveConf.setBoolVar(METASTORE_VALIDATE_CONSTRAINTS, true);
hiveConf.setBoolVar(METASTORE_VALIDATE_COLUMNS, true);
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkStandaloneHiveRunner.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkStandaloneHiveRunner.java
index 5caab72..b505085 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkStandaloneHiveRunner.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkStandaloneHiveRunner.java
@@ -19,38 +19,16 @@
package org.apache.flink.connectors.hive;
import org.apache.flink.table.catalog.hive.HiveTestUtils;
-import org.apache.flink.util.Preconditions;
import org.apache.flink.shaded.guava18.com.google.common.base.Joiner;
import org.apache.flink.shaded.guava18.com.google.common.base.Throwables;
-import org.apache.flink.shaded.guava18.com.google.common.io.Resources;
-import com.klarna.hiverunner.HiveServerContainer;
import com.klarna.hiverunner.HiveServerContext;
-import com.klarna.hiverunner.HiveShell;
-import com.klarna.hiverunner.HiveShellContainer;
-import com.klarna.hiverunner.annotations.HiveProperties;
-import com.klarna.hiverunner.annotations.HiveResource;
-import com.klarna.hiverunner.annotations.HiveRunnerSetup;
-import com.klarna.hiverunner.annotations.HiveSQL;
-import com.klarna.hiverunner.annotations.HiveSetupScript;
-import com.klarna.hiverunner.builder.HiveShellBuilder;
-import com.klarna.hiverunner.config.HiveRunnerConfig;
-import com.klarna.reflection.ReflectionUtils;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.HiveMetaStore;
-import org.junit.Ignore;
-import org.junit.internal.AssumptionViolatedException;
-import org.junit.internal.runners.model.EachTestNotifier;
import org.junit.rules.ExternalResource;
-import org.junit.rules.TemporaryFolder;
import org.junit.rules.TestRule;
-import org.junit.runner.Description;
-import org.junit.runner.notification.RunNotifier;
-import org.junit.runners.BlockJUnit4ClassRunner;
-import org.junit.runners.model.FrameworkMethod;
import org.junit.runners.model.InitializationError;
-import org.junit.runners.model.Statement;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -59,21 +37,12 @@ import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
-import java.lang.reflect.Field;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
-import java.nio.charset.Charset;
-import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
@@ -82,22 +51,19 @@ import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVEHISTORYFILELOC;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_IN_TEST;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.LOCALSCRATCHDIR;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTORECONNECTURLKEY;
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTOREURIS;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTOREWAREHOUSE;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.SCRATCHDIR;
-import static org.reflections.ReflectionUtils.withAnnotation;
/**
- * JUnit 4 runner that runs hive sql on a HiveServer residing in this JVM. No external dependencies needed.
- * Inspired by StandaloneHiveRunner.java (almost copied), just using local meta store server instead of embedded
- * hive meta store.
+ * Hive runner that uses local standalone HMS instead of embedded. Currently no test requires a
+ * standalone HMS, but let's keep this around for a while just in case we need it in the future.
*/
-public class FlinkStandaloneHiveRunner extends BlockJUnit4ClassRunner {
+public class FlinkStandaloneHiveRunner extends FlinkEmbeddedHiveRunner {
private static final Logger LOGGER = LoggerFactory.getLogger(FlinkStandaloneHiveRunner.class);
private static final Duration HMS_START_TIMEOUT = Duration.ofSeconds(90);
private Future<Void> hmsWatcher;
private int hmsPort;
- private HiveShellContainer container;
- private HiveRunnerConfig config = new HiveRunnerConfig();
public FlinkStandaloneHiveRunner(Class<?> clazz) throws InitializationError {
super(clazz);
@@ -105,19 +71,16 @@ public class FlinkStandaloneHiveRunner extends BlockJUnit4ClassRunner {
@Override
protected List<TestRule> classRules() {
- final TemporaryFolder temporaryFolder = new TemporaryFolder();
try {
hmsPort = HiveTestUtils.getFreePort();
} catch (IOException e) {
throw new RuntimeException(e);
}
- HiveServerContext context = new FlinkStandaloneHiveServerContext(temporaryFolder, config, hmsPort);
List<TestRule> rules = super.classRules();
ExternalResource hms = new ExternalResource() {
@Override
protected void before() throws Throwable {
- LOGGER.info("Setting up {} in {}", getName(), temporaryFolder.getRoot().getAbsolutePath());
- hmsWatcher = startHMS(getTestClass().getJavaClass(), context, hmsPort);
+ hmsWatcher = startHMS(context, hmsPort);
}
@Override
@@ -127,243 +90,17 @@ public class FlinkStandaloneHiveRunner extends BlockJUnit4ClassRunner {
}
}
};
- ExternalResource hiveShell = new ExternalResource() {
- @Override
- protected void before() throws Throwable {
- container = createHiveServerContainer(getTestClass().getJavaClass(), context);
- }
-
- @Override
- protected void after() {
- tearDown();
- }
- };
- rules.add(hiveShell);
- rules.add(hms);
- rules.add(temporaryFolder);
+ // the hms needs to be added before the temporary folder
+ rules.add(rules.size() - 1, hms);
return rules;
}
- @Override
- protected void runChild(final FrameworkMethod method, RunNotifier notifier) {
- Description description = describeChild(method);
- if (method.getAnnotation(Ignore.class) != null) {
- notifier.fireTestIgnored(description);
- } else {
- EachTestNotifier eachNotifier = new EachTestNotifier(notifier, description);
- eachNotifier.fireTestStarted();
- try {
- runTestMethod(method, eachNotifier);
- } finally {
- eachNotifier.fireTestFinished();
- }
- }
- }
-
- /**
- * Runs a {@link Statement} that represents a leaf (aka atomic) test.
- */
- private void runTestMethod(FrameworkMethod method,
- EachTestNotifier notifier) {
- Statement statement = methodBlock(method);
-
- try {
- statement.evaluate();
- } catch (AssumptionViolatedException e) {
- notifier.addFailedAssumption(e);
- } catch (Throwable e) {
- notifier.addFailure(e);
- }
- }
-
- private void tearDown() {
- if (container != null) {
- LOGGER.info("Tearing down {}", getName());
- try {
- container.tearDown();
- } catch (Throwable e) {
- LOGGER.warn("Tear down failed: " + e.getMessage(), e);
- }
- }
- }
-
- /**
- * Traverses the test class annotations. Will inject a HiveShell in the test case that envelopes the HiveServer.
- */
- private HiveShellContainer createHiveServerContainer(final Class testClass, HiveServerContext context)
- throws Exception {
-
- final HiveServerContainer hiveServerContainer = new HiveServerContainer(context);
-
- HiveShellBuilder hiveShellBuilder = new HiveShellBuilder();
- HiveRunnerShim hiveRunnerShim = HiveRunnerShimLoader.load();
- hiveRunnerShim.setCommandShellEmulation(hiveShellBuilder, config);
-
- HiveShellField shellSetter = loadScriptsUnderTest(testClass, hiveShellBuilder);
-
- hiveShellBuilder.setHiveServerContainer(hiveServerContainer);
-
- loadAnnotatedResources(testClass, hiveShellBuilder);
-
- loadAnnotatedProperties(testClass, hiveShellBuilder);
-
- loadAnnotatedSetupScripts(testClass, hiveShellBuilder);
-
- // Build shell
- final HiveShellContainer shell = hiveShellBuilder.buildShell();
-
- // Set shell
- shellSetter.setShell(shell);
-
- if (shellSetter.isAutoStart()) {
- shell.start();
- }
-
- return shell;
- }
-
- private void loadAnnotatesHiveRunnerConfig(Class testClass) {
- Set<Field> fields = ReflectionUtils.getAllFields(testClass, withAnnotation(HiveRunnerSetup.class));
- Preconditions.checkState(fields.size() <= 1,
- "Exact one field of type HiveRunnerConfig should to be annotated with @HiveRunnerSetup");
-
- // Override the config with test case config. Taking care to not replace the config instance since it
- // has been passes around and referenced by some of the other test rules.
- if (!fields.isEmpty()) {
- Field field = fields.iterator().next();
- Preconditions.checkState(ReflectionUtils.isOfType(field, HiveRunnerConfig.class),
- "Field annotated with @HiveRunnerSetup should be of type HiveRunnerConfig");
- config.override(ReflectionUtils.getStaticFieldValue(testClass, field.getName(), HiveRunnerConfig.class));
- }
- }
-
- private HiveShellField loadScriptsUnderTest(final Class testClass, HiveShellBuilder hiveShellBuilder) {
- try {
- Set<Field> fields = ReflectionUtils.getAllFields(testClass, withAnnotation(HiveSQL.class));
-
- Preconditions.checkState(fields.size() == 1, "Exactly one field should to be annotated with @HiveSQL");
-
- final Field field = fields.iterator().next();
- List<Path> scripts = new ArrayList<>();
- HiveSQL annotation = field.getAnnotation(HiveSQL.class);
- for (String scriptFilePath : annotation.files()) {
- Path file = Paths.get(Resources.getResource(scriptFilePath).toURI());
- Preconditions.checkState(Files.exists(file), "File " + file + " does not exist");
- scripts.add(file);
- }
-
- Charset charset = annotation.encoding().equals("") ?
- Charset.defaultCharset() : Charset.forName(annotation.encoding());
-
- final boolean isAutoStart = annotation.autoStart();
-
- hiveShellBuilder.setScriptsUnderTest(scripts, charset);
-
- return new HiveShellField() {
- @Override
- public void setShell(HiveShell shell) {
- ReflectionUtils.setStaticField(testClass, field.getName(), shell);
- }
-
- @Override
- public boolean isAutoStart() {
- return isAutoStart;
- }
- };
- } catch (Throwable t) {
- throw new IllegalArgumentException("Failed to init field annotated with @HiveSQL: " + t.getMessage(), t);
- }
- }
-
- private void loadAnnotatedSetupScripts(Class testClass, HiveShellBuilder hiveShellBuilder) {
- Set<Field> setupScriptFields = ReflectionUtils.getAllFields(testClass, withAnnotation(HiveSetupScript.class));
- for (Field setupScriptField : setupScriptFields) {
- if (ReflectionUtils.isOfType(setupScriptField, String.class)) {
- String script = ReflectionUtils.getStaticFieldValue(testClass, setupScriptField.getName(), String.class);
- hiveShellBuilder.addSetupScript(script);
- } else if (ReflectionUtils.isOfType(setupScriptField, File.class) ||
- ReflectionUtils.isOfType(setupScriptField, Path.class)) {
- Path path = getMandatoryPathFromField(testClass, setupScriptField);
- hiveShellBuilder.addSetupScript(readAll(path));
- } else {
- throw new IllegalArgumentException(
- "Field annotated with @HiveSetupScript currently only supports type String, File and Path");
- }
- }
- }
-
- private static String readAll(Path path) {
- try {
- return new String(Files.readAllBytes(path), StandardCharsets.UTF_8);
- } catch (IOException e) {
- throw new IllegalStateException("Unable to read " + path + ": " + e.getMessage(), e);
- }
- }
-
- private void loadAnnotatedResources(Class testClass, HiveShellBuilder workFlowBuilder) throws IOException {
- Set<Field> fields = ReflectionUtils.getAllFields(testClass, withAnnotation(HiveResource.class));
-
- for (Field resourceField : fields) {
-
- HiveResource annotation = resourceField.getAnnotation(HiveResource.class);
- String targetFile = annotation.targetFile();
-
- if (ReflectionUtils.isOfType(resourceField, String.class)) {
- String data = ReflectionUtils.getStaticFieldValue(testClass, resourceField.getName(), String.class);
- workFlowBuilder.addResource(targetFile, data);
- } else if (ReflectionUtils.isOfType(resourceField, File.class) ||
- ReflectionUtils.isOfType(resourceField, Path.class)) {
- Path dataFile = getMandatoryPathFromField(testClass, resourceField);
- workFlowBuilder.addResource(targetFile, dataFile);
- } else {
- throw new IllegalArgumentException(
- "Fields annotated with @HiveResource currently only supports field type String, File or Path");
- }
- }
- }
-
- private Path getMandatoryPathFromField(Class testClass, Field resourceField) {
- Path path;
- if (ReflectionUtils.isOfType(resourceField, File.class)) {
- File dataFile = ReflectionUtils.getStaticFieldValue(testClass, resourceField.getName(), File.class);
- path = Paths.get(dataFile.toURI());
- } else if (ReflectionUtils.isOfType(resourceField, Path.class)) {
- path = ReflectionUtils.getStaticFieldValue(testClass, resourceField.getName(), Path.class);
- } else {
- throw new IllegalArgumentException(
- "Only Path or File type is allowed on annotated field " + resourceField);
- }
-
- Preconditions.checkArgument(Files.exists(path), "File %s does not exist", path);
- return path;
- }
-
- private void loadAnnotatedProperties(Class testClass, HiveShellBuilder workFlowBuilder) {
- for (Field hivePropertyField : ReflectionUtils.getAllFields(testClass, withAnnotation(HiveProperties.class))) {
- Preconditions.checkState(ReflectionUtils.isOfType(hivePropertyField, Map.class),
- "Field annotated with @HiveProperties should be of type Map<String, String>");
- workFlowBuilder.putAllProperties(
- ReflectionUtils.getStaticFieldValue(testClass, hivePropertyField.getName(), Map.class));
- }
- }
-
- /**
- * Used as a handle for the HiveShell field in the test case so that we may set it once the
- * HiveShell has been instantiated.
- */
- interface HiveShellField {
- void setShell(HiveShell shell);
-
- boolean isAutoStart();
- }
-
/**
* Launches HMS process and returns a Future representing that process.
*/
- private Future<Void> startHMS(Class testClass, HiveServerContext context, int port) throws Exception {
- // need to load hive runner config before the context is inited
- loadAnnotatesHiveRunnerConfig(testClass);
+ private static Future<Void> startHMS(HiveServerContext context, int port) throws Exception {
context.init();
+ context.getHiveConf().setVar(METASTOREURIS, "thrift://localhost:" + port);
HiveConf outsideConf = context.getHiveConf();
List<String> args = new ArrayList<>();
String javaHome = System.getProperty("java.home");
@@ -382,8 +119,7 @@ public class FlinkStandaloneHiveRunner extends BlockJUnit4ClassRunner {
String.valueOf(outsideConf.getBoolean("hive.warehouse.subdir.inherit.perms", true))));
args.add(hiveCmdLineConfig("hadoop.tmp.dir", outsideConf.get("hadoop.tmp.dir")));
args.add(hiveCmdLineConfig("test.log.dir", outsideConf.get("test.log.dir")));
- String metaStorageUrl = "jdbc:derby:memory:" + UUID.randomUUID().toString();
- args.add(hiveCmdLineConfig(METASTORECONNECTURLKEY.varname, metaStorageUrl + ";create=true"));
+ args.add(hiveCmdLineConfig(METASTORECONNECTURLKEY.varname, outsideConf.getVar(METASTORECONNECTURLKEY)));
// config derby.log file
File derbyLog = File.createTempFile("derby", ".log");
derbyLog.deleteOnExit();
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveRunnerITCase.java
similarity index 58%
copy from flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorITCase.java
copy to flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveRunnerITCase.java
index 8bc5eb1..b5a2055 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorITCase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveRunnerITCase.java
@@ -23,16 +23,11 @@ import org.apache.flink.table.HiveVersionTestUtil;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.api.constraints.UniqueConstraint;
-import org.apache.flink.table.api.internal.TableImpl;
-import org.apache.flink.table.catalog.CatalogBaseTable;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.flink.table.catalog.hive.HiveTestUtils;
-import org.apache.flink.table.catalog.hive.client.HiveMetastoreClientFactory;
-import org.apache.flink.table.catalog.hive.client.HiveMetastoreClientWrapper;
import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
+import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
import org.apache.flink.types.Row;
import org.apache.flink.util.ArrayUtils;
import org.apache.flink.util.CollectionUtil;
@@ -41,16 +36,15 @@ import com.klarna.hiverunner.HiveShell;
import com.klarna.hiverunner.annotations.HiveRunnerSetup;
import com.klarna.hiverunner.annotations.HiveSQL;
import com.klarna.hiverunner.config.HiveRunnerConfig;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.ql.lockmgr.DbTxnManager;
+import org.junit.AfterClass;
import org.junit.Assume;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
+import java.io.IOException;
import java.sql.Date;
import java.sql.Timestamp;
import java.time.LocalDate;
@@ -62,6 +56,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -69,14 +64,14 @@ import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_IN_TEST;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_TXN_MANAGER;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
/**
- * Test hive connector with table API.
+ * Tests that need to run with hive runner. Since hive runner is heavy, make sure to add test cases here only if a
+ * test requires hive-side functionalities, e.g. preparing or validating data in hive.
*/
-@RunWith(FlinkStandaloneHiveRunner.class)
-public class TableEnvHiveConnectorITCase {
+@RunWith(FlinkEmbeddedHiveRunner.class)
+public class HiveRunnerITCase {
@HiveSQL(files = {})
private static HiveShell hiveShell;
@@ -93,51 +88,134 @@ public class TableEnvHiveConnectorITCase {
}};
private static HiveCatalog hiveCatalog;
- private static HiveMetastoreClientWrapper hmsClient;
@BeforeClass
- public static void setup() {
+ public static void createCatalog() throws IOException {
HiveConf hiveConf = hiveShell.getHiveConf();
hiveCatalog = HiveTestUtils.createHiveCatalog(hiveConf);
hiveCatalog.open();
- hmsClient = HiveMetastoreClientFactory.create(hiveConf, HiveShimLoader.getHiveVersion());
+ }
+
+ @AfterClass
+ public static void closeCatalog() {
+ if (hiveCatalog != null) {
+ hiveCatalog.close();
+ }
}
@Test
- public void testDefaultPartitionName() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- tableEnv.executeSql("create table db1.src (x int, y int)");
- tableEnv.executeSql("create table db1.part (x int) partitioned by (y int)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src").addRow(new Object[]{1, 1}).addRow(new Object[]{2, null}).commit();
+ public void testInsertIntoNonPartitionTable() throws Exception {
+ List<Row> toWrite = generateRecords(5);
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(toWrite);
+ TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithHiveCatalog(hiveCatalog);
+ tableEnv.executeSql("create table default_catalog.default_database.src (i int,l bigint,d double,s string) " +
+ "with ('connector'='COLLECTION','is-bounded' = 'true')");
+
+ tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
+ tableEnv.executeSql("create table dest (i int,l bigint,d double,s string)");
+ try {
+ tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
+ tableEnv.executeSql("insert into dest select * from default_catalog.default_database.src").await();
+ verifyWrittenData(toWrite, hiveShell.executeQuery("select * from dest"));
+ } finally {
+ tableEnv.executeSql("drop table dest");
+ }
+ }
- // test generating partitions with default name
- tableEnv.executeSql("insert into db1.part select * from db1.src").await();
- HiveConf hiveConf = hiveShell.getHiveConf();
- String defaultPartName = hiveConf.getVar(HiveConf.ConfVars.DEFAULTPARTITIONNAME);
- Table hiveTable = hmsClient.getTable("db1", "part");
- Path defaultPartPath = new Path(hiveTable.getSd().getLocation(), "y=" + defaultPartName);
- FileSystem fs = defaultPartPath.getFileSystem(hiveConf);
- assertTrue(fs.exists(defaultPartPath));
+ @Test
+ public void testWriteComplexType() throws Exception {
+ TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithHiveCatalog(hiveCatalog);
+
+ Row row = new Row(3);
+ Object[] array = new Object[]{1, 2, 3};
+ Map<Integer, String> map = new HashMap<>();
+ map.put(1, "a");
+ map.put(2, "b");
+ Row struct = new Row(2);
+ struct.setField(0, 3);
+ struct.setField(1, "c");
+ row.setField(0, array);
+ row.setField(1, map);
+ row.setField(2, struct);
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(Collections.singletonList(row));
+ tableEnv.executeSql("create table default_catalog.default_database.complexSrc (a array<int>,m map<int, string>,s row<f1 int,f2 string>) " +
+ "with ('connector'='COLLECTION','is-bounded' = 'true')");
+
+ tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
+ tableEnv.executeSql("create table dest (a array<int>,m map<int, string>,s struct<f1:int,f2:string>)");
- TableImpl flinkTable = (TableImpl) tableEnv.sqlQuery("select y, x from db1.part order by x");
- List<Row> rows = CollectionUtil.iteratorToList(flinkTable.execute().collect());
- assertEquals(Arrays.toString(new String[]{"1,1", "null,2"}), rows.toString());
+ try {
+ tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
+ tableEnv.executeSql("insert into dest select * from default_catalog.default_database.complexSrc").await();
+ List<String> result = hiveShell.executeQuery("select * from dest");
+ assertEquals(1, result.size());
+ assertEquals("[1,2,3]\t{1:\"a\",2:\"b\"}\t{\"f1\":3,\"f2\":\"c\"}", result.get(0));
+ } finally {
+ tableEnv.executeSql("drop table dest");
+ }
+ }
- tableEnv.executeSql("drop database db1 cascade");
+ @Test
+ public void testWriteNestedComplexType() throws Exception {
+
+ TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithHiveCatalog(hiveCatalog);
+
+ Row row = new Row(1);
+ Object[] array = new Object[3];
+ row.setField(0, array);
+ for (int i = 0; i < array.length; i++) {
+ Row struct = new Row(2);
+ struct.setField(0, 1 + i);
+ struct.setField(1, String.valueOf((char) ('a' + i)));
+ array[i] = struct;
+ }
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(Collections.singletonList(row));
+ tableEnv.executeSql("create table default_catalog.default_database.nestedSrc (a array<row<f1 int,f2 string>>) " +
+ "with ('connector'='COLLECTION','is-bounded' = 'true')");
+
+ tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
+ tableEnv.executeSql("create table dest (a array<struct<f1:int,f2:string>>)");
+
+ try {
+ tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
+ tableEnv.executeSql("insert into dest select * from default_catalog.default_database.nestedSrc").await();
+ List<String> result = hiveShell.executeQuery("select * from dest");
+ assertEquals(1, result.size());
+ assertEquals("[{\"f1\":1,\"f2\":\"a\"},{\"f1\":2,\"f2\":\"b\"},{\"f1\":3,\"f2\":\"c\"}]", result.get(0));
+ } finally {
+ tableEnv.executeSql("drop table dest");
+ }
}
@Test
- public void testGetNonExistingFunction() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
+ public void testWriteNullValues() throws Exception {
+ TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode(SqlDialect.HIVE);
+ tableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
+ tableEnv.useCatalog(hiveCatalog.getName());
tableEnv.executeSql("create database db1");
- tableEnv.executeSql("create table db1.src (d double, s string)");
- tableEnv.executeSql("create table db1.dest (x bigint)");
-
- // just make sure the query runs through, no need to verify result
- tableEnv.executeSql("insert into db1.dest select count(d) from db1.src").await();
+ try {
+ // 17 data types
+ tableEnv.executeSql("create table db1.src" +
+ "(t tinyint,s smallint,i int,b bigint,f float,d double,de decimal(10,5),ts timestamp,dt date," +
+ "str string,ch char(5),vch varchar(8),bl boolean,bin binary,arr array<int>,mp map<int,string>,strt struct<f1:int,f2:string>)");
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "src")
+ .addRow(new Object[]{null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null})
+ .commit();
+ hiveShell.execute("create table db1.dest like db1.src");
- tableEnv.executeSql("drop database db1 cascade");
+ tableEnv.executeSql("insert into db1.dest select * from db1.src").await();
+ List<String> results = hiveShell.executeQuery("select * from db1.dest");
+ assertEquals(1, results.size());
+ String[] cols = results.get(0).split("\t");
+ assertEquals(17, cols.length);
+ assertEquals("NULL", cols[0]);
+ assertEquals(1, new HashSet<>(Arrays.asList(cols)).size());
+ } finally {
+ tableEnv.executeSql("drop database db1 cascade");
+ }
}
@Test
@@ -152,77 +230,6 @@ public class TableEnvHiveConnectorITCase {
}
}
- private void readWriteFormat(String format) throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
-
- tableEnv.executeSql("create database db1");
-
- // create source and dest tables
- String suffix;
- if (format.equals("csv")) {
- suffix = "row format serde 'org.apache.hadoop.hive.serde2.OpenCSVSerde'";
- } else {
- suffix = "stored as " + format;
- }
- String tableSchema;
- // use 2018-08-20 00:00:00.1 to avoid multi-version print difference.
- List<Object> row1 = new ArrayList<>(Arrays.asList(1, "a", "2018-08-20 00:00:00.1"));
- List<Object> row2 = new ArrayList<>(Arrays.asList(2, "b", "2019-08-26 00:00:00.1"));
- // some data types are not supported for parquet tables in early versions -- https://issues.apache.org/jira/browse/HIVE-6384
- if (HiveVersionTestUtil.HIVE_120_OR_LATER || !format.equals("parquet")) {
- tableSchema = "(i int,s string,ts timestamp,dt date)";
- row1.add("2018-08-20");
- row2.add("2019-08-26");
- } else {
- tableSchema = "(i int,s string,ts timestamp)";
- }
-
- tableEnv.executeSql(String.format(
- "create table db1.src %s partitioned by (p1 string, p2 timestamp) %s", tableSchema, suffix));
- tableEnv.executeSql(String.format(
- "create table db1.dest %s partitioned by (p1 string, p2 timestamp) %s", tableSchema, suffix));
-
- // prepare source data with Hive
- // TABLE keyword in INSERT INTO is mandatory prior to 1.1.0
- hiveShell.execute(String.format(
- "insert into table db1.src partition(p1='first',p2='2018-08-20 00:00:00.1') values (%s)",
- toRowValue(row1)));
- hiveShell.execute(String.format(
- "insert into table db1.src partition(p1='second',p2='2018-08-26 00:00:00.1') values (%s)",
- toRowValue(row2)));
-
- List<String> expected = Arrays.asList(
- String.join("\t", ArrayUtils.concat(
- row1.stream().map(Object::toString).toArray(String[]::new),
- new String[]{"first", "2018-08-20 00:00:00.1"})),
- String.join("\t", ArrayUtils.concat(
- row2.stream().map(Object::toString).toArray(String[]::new),
- new String[]{"second", "2018-08-26 00:00:00.1"})));
-
- verifyFlinkQueryResult(tableEnv.sqlQuery("select * from db1.src"), expected);
-
- // Ignore orc write test for Hive version 2.0.x for now due to FLINK-13998
- if (!format.equals("orc") || !HiveShimLoader.getHiveVersion().startsWith("2.0")) {
- // populate dest table with source table
- tableEnv.executeSql("insert into db1.dest select * from db1.src").await();
-
- // verify data on hive side
- verifyHiveQueryResult("select * from db1.dest", expected);
- }
-
- tableEnv.executeSql("drop database db1 cascade");
- }
-
- private String toRowValue(List<Object> row) {
- return row.stream().map(o -> {
- String res = o.toString();
- if (o instanceof String) {
- res = "'" + res + "'";
- }
- return res;
- }).collect(Collectors.joining(","));
- }
-
@Test
public void testDecimal() throws Exception {
TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
@@ -257,7 +264,7 @@ public class TableEnvHiveConnectorITCase {
try {
// non-partitioned
tableEnv.executeSql("create table db1.dest (x int, y string)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "dest").addRow(new Object[]{1, "a"}).addRow(new Object[]{2, "b"}).commit();
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "dest").addRow(new Object[]{1, "a"}).addRow(new Object[]{2, "b"}).commit();
verifyHiveQueryResult("select * from db1.dest", Arrays.asList("1\ta", "2\tb"));
tableEnv.executeSql("insert overwrite db1.dest values (3, 'c')").await();
@@ -265,8 +272,8 @@ public class TableEnvHiveConnectorITCase {
// static partition
tableEnv.executeSql("create table db1.part(x int) partitioned by (y int)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part").addRow(new Object[]{1}).commit("y=1");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part").addRow(new Object[]{2}).commit("y=2");
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "part").addRow(new Object[]{1}).commit("y=1");
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "part").addRow(new Object[]{2}).commit("y=2");
tableEnv = getTableEnvWithHiveCatalog();
tableEnv.executeSql("insert overwrite db1.part partition (y=1) select 100").await();
verifyHiveQueryResult("select * from db1.part", Arrays.asList("100\t1", "2\t2"));
@@ -287,7 +294,7 @@ public class TableEnvHiveConnectorITCase {
tableEnv.executeSql("create database db1");
try {
tableEnv.executeSql("create table db1.src (x int)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src").addRow(new Object[]{1}).addRow(new Object[]{2}).commit();
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "src").addRow(new Object[]{1}).addRow(new Object[]{2}).commit();
tableEnv.executeSql("create table db1.dest (x int) partitioned by (p1 string, p2 double)");
tableEnv.executeSql("insert into db1.dest partition (p1='1''1', p2=1.1) select x from db1.src").await();
assertEquals(1, hiveCatalog.listPartitions(new ObjectPath("db1", "dest")).size());
@@ -303,7 +310,7 @@ public class TableEnvHiveConnectorITCase {
tableEnv.executeSql("create database db1");
try {
tableEnv.executeSql("create table db1.src (x int, y string, z double)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "src")
.addRow(new Object[]{1, "a", 1.1})
.addRow(new Object[]{2, "a", 2.2})
.addRow(new Object[]{3, "b", 3.3})
@@ -323,7 +330,7 @@ public class TableEnvHiveConnectorITCase {
tableEnv.executeSql("create database db1");
try {
tableEnv.executeSql("create table db1.src (x int, y string)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src").addRow(new Object[]{1, "a"}).addRow(new Object[]{2, "b"}).commit();
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "src").addRow(new Object[]{1, "a"}).addRow(new Object[]{2, "b"}).commit();
tableEnv.executeSql("create table db1.dest (x int) partitioned by (p1 double, p2 string)");
tableEnv.executeSql("insert into db1.dest partition (p1=1.1) select x,y from db1.src").await();
assertEquals(2, hiveCatalog.listPartitions(new ObjectPath("db1", "dest")).size());
@@ -334,127 +341,13 @@ public class TableEnvHiveConnectorITCase {
}
@Test
- public void testDateTimestampPartitionColumns() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.part(x int) partitioned by (dt date,ts timestamp)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part")
- .addRow(new Object[]{1})
- .addRow(new Object[]{2})
- .commit("dt='2019-12-23',ts='2019-12-23 00:00:00'");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part")
- .addRow(new Object[]{3})
- .commit("dt='2019-12-25',ts='2019-12-25 16:23:43.012'");
- List<Row> results = CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.part order by x").execute().collect());
- assertEquals("[1,2019-12-23,2019-12-23T00:00, 2,2019-12-23,2019-12-23T00:00, 3,2019-12-25,2019-12-25T16:23:43.012]", results.toString());
-
- results = CollectionUtil.iteratorToList(tableEnv.sqlQuery("select x from db1.part where dt=cast('2019-12-25' as date)").execute().collect());
- assertEquals("[3]", results.toString());
-
- tableEnv.executeSql("insert into db1.part select 4,cast('2019-12-31' as date),cast('2019-12-31 12:00:00.0' as timestamp)")
- .await();
- results = CollectionUtil.iteratorToList(tableEnv.sqlQuery("select max(dt) from db1.part").execute().collect());
- assertEquals("[2019-12-31]", results.toString());
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- @Test
- public void testUDTF() {
- // W/o https://issues.apache.org/jira/browse/HIVE-11878 Hive registers the App classloader as the classloader
- // for the UDTF and closes the App classloader when we tear down the session. This causes problems for JUnit code
- // and shutdown hooks that have to run after the test finishes, because App classloader can no longer load new
- // classes. And will crash the forked JVM, thus failing the test phase.
- // Therefore disable such tests for older Hive versions.
- String hiveVersion = HiveShimLoader.getHiveVersion();
- Assume.assumeTrue(hiveVersion.compareTo("2.0.0") >= 0 || hiveVersion.compareTo("1.3.0") >= 0);
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.simple (i int,a array<int>)");
- tableEnv.executeSql("create table db1.nested (a array<map<int, string>>)");
- tableEnv.executeSql("create function hiveudtf as 'org.apache.hadoop.hive.ql.udf.generic.GenericUDTFExplode'");
- hiveShell.insertInto("db1", "simple").addRow(3, Arrays.asList(1, 2, 3)).commit();
- Map<Integer, String> map1 = new HashMap<>();
- map1.put(1, "a");
- map1.put(2, "b");
- Map<Integer, String> map2 = new HashMap<>();
- map2.put(3, "c");
- hiveShell.insertInto("db1", "nested").addRow(Arrays.asList(map1, map2)).commit();
-
- List<Row> results = CollectionUtil.iteratorToList(
- tableEnv.sqlQuery("select x from db1.simple, lateral table(hiveudtf(a)) as T(x)").execute().collect());
- assertEquals("[1, 2, 3]", results.toString());
- results = CollectionUtil.iteratorToList(
- tableEnv.sqlQuery("select x from db1.nested, lateral table(hiveudtf(a)) as T(x)").execute().collect());
- assertEquals("[{1=a, 2=b}, {3=c}]", results.toString());
-
- tableEnv.executeSql("create table db1.ts (a array<timestamp>)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "ts").addRow(new Object[]{
- new Object[]{Timestamp.valueOf("2015-04-28 15:23:00"), Timestamp.valueOf("2016-06-03 17:05:52")}})
- .commit();
- results = CollectionUtil.iteratorToList(
- tableEnv.sqlQuery("select x from db1.ts, lateral table(hiveudtf(a)) as T(x)").execute().collect());
- assertEquals("[2015-04-28T15:23, 2016-06-03T17:05:52]", results.toString());
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- tableEnv.executeSql("drop function hiveudtf");
- }
- }
-
- @Test
- public void testNotNullConstraints() throws Exception {
- Assume.assumeTrue(HiveVersionTestUtil.HIVE_310_OR_LATER);
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.tbl (x int,y bigint not null enable rely,z string not null enable norely)");
- CatalogBaseTable catalogTable = hiveCatalog.getTable(new ObjectPath("db1", "tbl"));
- TableSchema tableSchema = catalogTable.getSchema();
- assertTrue("By default columns should be nullable",
- tableSchema.getFieldDataTypes()[0].getLogicalType().isNullable());
- assertFalse("NOT NULL columns should be reflected in table schema",
- tableSchema.getFieldDataTypes()[1].getLogicalType().isNullable());
- assertTrue("NOT NULL NORELY columns should be considered nullable",
- tableSchema.getFieldDataTypes()[2].getLogicalType().isNullable());
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
+ public void testBatchCompressTextTable() throws Exception {
+ testCompressTextTable(true);
}
@Test
- public void testPKConstraint() throws Exception {
- // While PK constraints are supported since Hive 2.1.0, the constraints cannot be RELY in 2.x versions.
- // So let's only test for 3.x.
- Assume.assumeTrue(HiveVersionTestUtil.HIVE_310_OR_LATER);
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- // test rely PK constraints
- tableEnv.executeSql("create table db1.tbl1 (x tinyint,y smallint,z int, primary key (x,z) disable novalidate rely)");
- CatalogBaseTable catalogTable = hiveCatalog.getTable(new ObjectPath("db1", "tbl1"));
- TableSchema tableSchema = catalogTable.getSchema();
- assertTrue(tableSchema.getPrimaryKey().isPresent());
- UniqueConstraint pk = tableSchema.getPrimaryKey().get();
- assertEquals(2, pk.getColumns().size());
- assertTrue(pk.getColumns().containsAll(Arrays.asList("x", "z")));
-
- // test norely PK constraints
- tableEnv.executeSql("create table db1.tbl2 (x tinyint,y smallint, primary key (x) disable norely)");
- catalogTable = hiveCatalog.getTable(new ObjectPath("db1", "tbl2"));
- tableSchema = catalogTable.getSchema();
- assertFalse(tableSchema.getPrimaryKey().isPresent());
-
- // test table w/o PK
- tableEnv.executeSql("create table db1.tbl3 (x tinyint)");
- catalogTable = hiveCatalog.getTable(new ObjectPath("db1", "tbl3"));
- tableSchema = catalogTable.getSchema();
- assertFalse(tableSchema.getPrimaryKey().isPresent());
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
+ public void testStreamCompressTextTable() throws Exception {
+ testCompressTextTable(false);
}
@Test
@@ -464,7 +357,7 @@ public class TableEnvHiveConnectorITCase {
try {
tableEnv.executeSql("create table db1.src (ts timestamp)");
tableEnv.executeSql("create table db1.dest (ts timestamp)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "src")
.addRow(new Object[]{Timestamp.valueOf("2019-11-11 00:00:00")})
.addRow(new Object[]{Timestamp.valueOf("2019-12-03 15:43:32.123456789")})
.commit();
@@ -488,7 +381,7 @@ public class TableEnvHiveConnectorITCase {
try {
tableEnv.executeSql("create table db1.src (dt date)");
tableEnv.executeSql("create table db1.dest (dt date)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "src")
.addRow(new Object[]{Date.valueOf("2019-12-09")})
.addRow(new Object[]{Date.valueOf("2019-12-12")})
.commit();
@@ -506,12 +399,12 @@ public class TableEnvHiveConnectorITCase {
}
@Test
- public void testViews() {
+ public void testViews() throws Exception {
TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
tableEnv.executeSql("create database db1");
try {
tableEnv.executeSql("create table db1.src (key int,val string)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "src")
.addRow(new Object[]{1, "a"})
.addRow(new Object[]{1, "aa"})
.addRow(new Object[]{1, "aaa"})
@@ -520,7 +413,7 @@ public class TableEnvHiveConnectorITCase {
.addRow(new Object[]{3, "ccc"})
.commit();
tableEnv.executeSql("create table db1.keys (key int,name string)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "keys")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "keys")
.addRow(new Object[]{1, "key1"})
.addRow(new Object[]{2, "key2"})
.addRow(new Object[]{3, "key3"})
@@ -556,42 +449,43 @@ public class TableEnvHiveConnectorITCase {
}
}
- private void testCompressTextTable(boolean batch) throws Exception {
- TableEnvironment tableEnv = batch ?
- getTableEnvWithHiveCatalog() :
- getStreamTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.src (x string,y string)");
- hiveShell.execute("create table db1.dest like db1.src");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src")
- .addRow(new Object[]{"a", "b"})
- .addRow(new Object[]{"c", "d"})
- .commit();
- hiveCatalog.getHiveConf().setBoolVar(HiveConf.ConfVars.COMPRESSRESULT, true);
- tableEnv.executeSql("insert into db1.dest select * from db1.src").await();
- List<String> expected = Arrays.asList("a\tb", "c\td");
- verifyHiveQueryResult("select * from db1.dest", expected);
- verifyFlinkQueryResult(tableEnv.sqlQuery("select * from db1.dest"), expected);
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
+ @Test
+ public void testBatchTransactionalTable() {
+ testTransactionalTable(true);
}
@Test
- public void testBatchCompressTextTable() throws Exception {
- testCompressTextTable(true);
+ public void testStreamTransactionalTable() {
+ testTransactionalTable(false);
}
@Test
- public void testStreamCompressTextTable() throws Exception {
- testCompressTextTable(false);
+ public void testOrcSchemaEvol() throws Exception {
+ // not supported until 2.1.0 -- https://issues.apache.org/jira/browse/HIVE-11981,
+ // https://issues.apache.org/jira/browse/HIVE-13178
+ Assume.assumeTrue(HiveVersionTestUtil.HIVE_210_OR_LATER);
+ TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
+ tableEnv.executeSql("create database db1");
+ try {
+ tableEnv.executeSql("create table db1.src (x smallint,y int) stored as orc");
+ hiveShell.execute("insert into table db1.src values (1,100),(2,200)");
+
+ tableEnv.getConfig().getConfiguration().setBoolean(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER, true);
+
+ tableEnv.executeSql("alter table db1.src change x x int");
+ assertEquals("[1,100, 2,200]", CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.src").execute().collect()).toString());
+
+ tableEnv.executeSql("alter table db1.src change y y string");
+ assertEquals("[1,100, 2,200]", CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.src").execute().collect()).toString());
+ } finally {
+ tableEnv.executeSql("drop database db1 cascade");
+ }
}
private void testTransactionalTable(boolean batch) {
TableEnvironment tableEnv = batch ?
- getTableEnvWithHiveCatalog() :
- getStreamTableEnvWithHiveCatalog();
+ getTableEnvWithHiveCatalog() :
+ getStreamTableEnvWithHiveCatalog();
tableEnv.executeSql("create database db1");
try {
tableEnv.executeSql("create table db1.src (x string,y string)");
@@ -617,161 +511,134 @@ public class TableEnvHiveConnectorITCase {
}
}
- @Test
- public void testBatchTransactionalTable() {
- testTransactionalTable(true);
- }
-
- @Test
- public void testStreamTransactionalTable() {
- testTransactionalTable(false);
- }
-
- @Test
- public void testRegexSerDe() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
+ private void testCompressTextTable(boolean batch) throws Exception {
+ TableEnvironment tableEnv = batch ?
+ getTableEnvWithHiveCatalog() :
+ getStreamTableEnvWithHiveCatalog();
tableEnv.executeSql("create database db1");
try {
- tableEnv.executeSql("create table db1.src (x int,y string) " +
- "row format serde 'org.apache.hadoop.hive.serde2.RegexSerDe' " +
- "with serdeproperties ('input.regex'='([\\\\d]+)\\u0001([\\\\S]+)')");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src")
- .addRow(new Object[]{1, "a"})
- .addRow(new Object[]{2, "ab"})
+ tableEnv.executeSql("create table db1.src (x string,y string)");
+ hiveShell.execute("create table db1.dest like db1.src");
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "src")
+ .addRow(new Object[]{"a", "b"})
+ .addRow(new Object[]{"c", "d"})
.commit();
- assertEquals("[1,a, 2,ab]", CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.src order by x").execute().collect()).toString());
+ hiveCatalog.getHiveConf().setBoolVar(HiveConf.ConfVars.COMPRESSRESULT, true);
+ tableEnv.executeSql("insert into db1.dest select * from db1.src").await();
+ List<String> expected = Arrays.asList("a\tb", "c\td");
+ verifyHiveQueryResult("select * from db1.dest", expected);
+ verifyFlinkQueryResult(tableEnv.sqlQuery("select * from db1.dest"), expected);
} finally {
tableEnv.executeSql("drop database db1 cascade");
}
}
- @Test
- public void testUpdatePartitionSD() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.dest (x int) partitioned by (p string) stored as rcfile");
- tableEnv.executeSql("insert overwrite db1.dest partition (p='1') select 1").await();
- tableEnv.executeSql("alter table db1.dest set fileformat sequencefile");
- tableEnv.executeSql("insert overwrite db1.dest partition (p='1') select 1").await();
- assertEquals("[1,1]", CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.dest").execute().collect()).toString());
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
+ private static TableEnvironment getTableEnvWithHiveCatalog() {
+ TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode(SqlDialect.HIVE);
+ tableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
+ tableEnv.useCatalog(hiveCatalog.getName());
+ return tableEnv;
}
- @Test
- public void testParquetNameMapping() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.t1 (x int,y int) stored as parquet");
- tableEnv.executeSql("insert into table db1.t1 values (1,10),(2,20)").await();
- Table hiveTable = hiveCatalog.getHiveTable(new ObjectPath("db1", "t1"));
- String location = hiveTable.getSd().getLocation();
- tableEnv.executeSql(String.format("create table db1.t2 (y int,x int) stored as parquet location '%s'", location));
- tableEnv.getConfig().getConfiguration().setBoolean(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER, true);
- assertEquals("[1, 2]", CollectionUtil.iteratorToList(tableEnv.sqlQuery("select x from db1.t1").execute().collect()).toString());
- assertEquals("[1, 2]", CollectionUtil.iteratorToList(tableEnv.sqlQuery("select x from db1.t2").execute().collect()).toString());
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
+ private TableEnvironment getStreamTableEnvWithHiveCatalog() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerStreamMode(env, SqlDialect.HIVE);
+ tableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
+ tableEnv.useCatalog(hiveCatalog.getName());
+ return tableEnv;
}
- @Test
- public void testOrcSchemaEvol() throws Exception {
- // not supported until 2.1.0 -- https://issues.apache.org/jira/browse/HIVE-11981,
- // https://issues.apache.org/jira/browse/HIVE-13178
- Assume.assumeTrue(HiveVersionTestUtil.HIVE_210_OR_LATER);
+ private void readWriteFormat(String format) throws Exception {
TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
+
tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.src (x smallint,y int) stored as orc");
- hiveShell.execute("insert into table db1.src values (1,100),(2,200)");
- tableEnv.getConfig().getConfiguration().setBoolean(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER, true);
+ // create source and dest tables
+ String suffix;
+ if (format.equals("csv")) {
+ suffix = "row format serde 'org.apache.hadoop.hive.serde2.OpenCSVSerde'";
+ } else {
+ suffix = "stored as " + format;
+ }
+ String tableSchema;
+ // use 2018-08-20 00:00:00.1 to avoid multi-version print difference.
+ List<Object> row1 = new ArrayList<>(Arrays.asList(1, "a", "2018-08-20 00:00:00.1"));
+ List<Object> row2 = new ArrayList<>(Arrays.asList(2, "b", "2019-08-26 00:00:00.1"));
+ // some data types are not supported for parquet tables in early versions -- https://issues.apache.org/jira/browse/HIVE-6384
+ if (HiveVersionTestUtil.HIVE_120_OR_LATER || !format.equals("parquet")) {
+ tableSchema = "(i int,s string,ts timestamp,dt date)";
+ row1.add("2018-08-20");
+ row2.add("2019-08-26");
+ } else {
+ tableSchema = "(i int,s string,ts timestamp)";
+ }
- tableEnv.executeSql("alter table db1.src change x x int");
- assertEquals("[1,100, 2,200]", CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.src").execute().collect()).toString());
+ tableEnv.executeSql(String.format(
+ "create table db1.src %s partitioned by (p1 string, p2 timestamp) %s", tableSchema, suffix));
+ tableEnv.executeSql(String.format(
+ "create table db1.dest %s partitioned by (p1 string, p2 timestamp) %s", tableSchema, suffix));
- tableEnv.executeSql("alter table db1.src change y y string");
- assertEquals("[1,100, 2,200]", CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.src").execute().collect()).toString());
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
+ // prepare source data with Hive
+ // TABLE keyword in INSERT INTO is mandatory prior to 1.1.0
+ hiveShell.execute(String.format(
+ "insert into table db1.src partition(p1='first',p2='2018-08-20 00:00:00.1') values (%s)",
+ toRowValue(row1)));
+ hiveShell.execute(String.format(
+ "insert into table db1.src partition(p1='second',p2='2018-08-26 00:00:00.1') values (%s)",
+ toRowValue(row2)));
- @Test
- public void testNonExistingPartitionFolder() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.part (x int) partitioned by (p int)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part").addRow(new Object[]{1}).commit("p=1");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part").addRow(new Object[]{2}).commit("p=2");
- tableEnv.executeSql("alter table db1.part add partition (p=3)");
- // remove one partition
- Path toRemove = new Path(hiveCatalog.getHiveTable(new ObjectPath("db1", "part")).getSd().getLocation(), "p=2");
- FileSystem fs = toRemove.getFileSystem(hiveShell.getHiveConf());
- fs.delete(toRemove, true);
-
- List<Row> results = CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.part").execute().collect());
- assertEquals("[1,1]", results.toString());
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
+ List<String> expected = Arrays.asList(
+ String.join("\t", ArrayUtils.concat(
+ row1.stream().map(Object::toString).toArray(String[]::new),
+ new String[]{"first", "2018-08-20 00:00:00.1"})),
+ String.join("\t", ArrayUtils.concat(
+ row2.stream().map(Object::toString).toArray(String[]::new),
+ new String[]{"second", "2018-08-26 00:00:00.1"})));
- @Test
- public void testInsertPartitionWithStarSource() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create table src (x int,y string)");
- HiveTestUtils.createTextTableInserter(
- hiveShell,
- "default",
- "src")
- .addRow(new Object[]{1, "a"})
- .commit();
- tableEnv.executeSql("create table dest (x int) partitioned by (p1 int,p2 string)");
- tableEnv.executeSql("insert into dest partition (p1=1) select * from src").await();
- List<Row> results = CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from dest").execute().collect());
- assertEquals("[1,1,a]", results.toString());
- tableEnv.executeSql("drop table if exists src");
- tableEnv.executeSql("drop table if exists dest");
- }
+ verifyFlinkQueryResult(tableEnv.sqlQuery("select * from db1.src"), expected);
- @Test
- public void testInsertPartitionWithValuesSource() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create table dest (x int) partitioned by (p1 int,p2 string)");
- tableEnv.executeSql("insert into dest partition (p1=1) values(1, 'a')").await();
- List<Row> results = CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from dest").execute().collect());
- assertEquals("[1,1,a]", results.toString());
- tableEnv.executeSql("drop table if exists dest");
- }
+ // Ignore orc write test for Hive version 2.0.x for now due to FLINK-13998
+ if (!format.equals("orc") || !HiveShimLoader.getHiveVersion().startsWith("2.0")) {
+ // populate dest table with source table
+ tableEnv.executeSql("insert into db1.dest select * from db1.src").await();
- private TableEnvironment getTableEnvWithHiveCatalog() {
- TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode(SqlDialect.HIVE);
- tableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
- tableEnv.useCatalog(hiveCatalog.getName());
- return tableEnv;
+ // verify data on hive side
+ verifyHiveQueryResult("select * from db1.dest", expected);
+ }
+
+ tableEnv.executeSql("drop database db1 cascade");
}
- private TableEnvironment getStreamTableEnvWithHiveCatalog() {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerStreamMode(env, SqlDialect.HIVE);
- tableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
- tableEnv.useCatalog(hiveCatalog.getName());
- return tableEnv;
+ private static void verifyWrittenData(List<Row> expected, List<String> results) throws Exception {
+ assertEquals(expected.size(), results.size());
+ Set<String> expectedSet = new HashSet<>();
+ for (int i = 0; i < results.size(); i++) {
+ expectedSet.add(expected.get(i).toString().replaceAll(",", "\t"));
+ }
+ assertEquals(expectedSet, new HashSet<>(results));
+ }
+
+ private static List<Row> generateRecords(int numRecords) {
+ int arity = 4;
+ List<Row> res = new ArrayList<>(numRecords);
+ for (int i = 0; i < numRecords; i++) {
+ Row row = new Row(arity);
+ row.setField(0, i);
+ row.setField(1, (long) i);
+ row.setField(2, Double.valueOf(String.valueOf(String.format("%d.%d", i, i))));
+ row.setField(3, String.valueOf((char) ('a' + i)));
+ res.add(row);
+ }
+ return res;
}
- private void verifyHiveQueryResult(String query, List<String> expected) {
+ private static void verifyHiveQueryResult(String query, List<String> expected) {
List<String> results = hiveShell.executeQuery(query);
assertEquals(expected.size(), results.size());
assertEquals(new HashSet<>(expected), new HashSet<>(results));
}
- private void verifyFlinkQueryResult(org.apache.flink.table.api.Table table, List<String> expected) throws Exception {
+ private static void verifyFlinkQueryResult(org.apache.flink.table.api.Table table, List<String> expected) throws Exception {
List<Row> rows = CollectionUtil.iteratorToList(table.execute().collect());
List<String> results = rows.stream().map(row ->
IntStream.range(0, row.getArity())
@@ -783,4 +650,14 @@ public class TableEnvHiveConnectorITCase {
assertEquals(expected.size(), results.size());
assertEquals(new HashSet<>(expected), new HashSet<>(results));
}
+
+ private static String toRowValue(List<Object> row) {
+ return row.stream().map(o -> {
+ String res = o.toString();
+ if (o instanceof String) {
+ res = "'" + res + "'";
+ }
+ return res;
+ }).collect(Collectors.joining(","));
+ }
}
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java
index 243e3a4..e7d64f7 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java
@@ -18,54 +18,32 @@
package org.apache.flink.connectors.hive;
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.io.InputFormat;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
-import org.apache.flink.api.java.io.CollectionInputFormat;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.util.FiniteTestSource;
-import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.SqlDialect;
-import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
-import org.apache.flink.table.catalog.CatalogTable;
-import org.apache.flink.table.catalog.CatalogTableImpl;
import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.catalog.config.CatalogConfig;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.flink.table.catalog.hive.HiveTestUtils;
-import org.apache.flink.table.sources.InputFormatTableSource;
-import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.utils.TypeConversions;
import org.apache.flink.types.Row;
import org.apache.flink.util.CollectionUtil;
-import com.klarna.hiverunner.HiveShell;
-import com.klarna.hiverunner.annotations.HiveSQL;
-import org.apache.hadoop.hive.conf.HiveConf;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
-import org.junit.runner.RunWith;
import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collection;
import java.util.Comparator;
-import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
-import java.util.Map;
-import java.util.Set;
import java.util.function.Consumer;
import static org.apache.flink.table.api.Expressions.$;
@@ -73,24 +51,17 @@ import static org.apache.flink.table.filesystem.FileSystemOptions.PARTITION_TIME
import static org.apache.flink.table.filesystem.FileSystemOptions.SINK_PARTITION_COMMIT_DELAY;
import static org.apache.flink.table.filesystem.FileSystemOptions.SINK_PARTITION_COMMIT_POLICY_KIND;
import static org.apache.flink.table.filesystem.FileSystemOptions.SINK_PARTITION_COMMIT_SUCCESS_FILE_NAME;
-import static org.junit.Assert.assertEquals;
/**
* Tests {@link HiveTableSink}.
*/
-@RunWith(FlinkStandaloneHiveRunner.class)
public class HiveTableSinkITCase {
- @HiveSQL(files = {})
- private static HiveShell hiveShell;
-
private static HiveCatalog hiveCatalog;
- private static HiveConf hiveConf;
@BeforeClass
public static void createCatalog() throws IOException {
- hiveConf = hiveShell.getHiveConf();
- hiveCatalog = HiveTestUtils.createHiveCatalog(hiveConf);
+ hiveCatalog = HiveTestUtils.createHiveCatalog();
hiveCatalog.open();
}
@@ -102,134 +73,6 @@ public class HiveTableSinkITCase {
}
@Test
- public void testInsertIntoNonPartitionTable() throws Exception {
- String dbName = "default";
- String tblName = "dest";
- RowTypeInfo rowTypeInfo = createHiveDestTable(dbName, tblName, 0);
- ObjectPath tablePath = new ObjectPath(dbName, tblName);
-
- TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode();
- List<Row> toWrite = generateRecords(5);
- Table src = tableEnv.fromTableSource(new CollectionTableSource(toWrite, rowTypeInfo));
- tableEnv.registerTable("src", src);
-
- tableEnv.registerCatalog("hive", hiveCatalog);
- tableEnv.sqlQuery("select * from src").executeInsert("hive.`default`.dest").await();
-
- verifyWrittenData(toWrite, hiveShell.executeQuery("select * from " + tblName));
-
- hiveCatalog.dropTable(tablePath, false);
- }
-
- @Test
- public void testWriteComplexType() throws Exception {
- String dbName = "default";
- String tblName = "dest";
- ObjectPath tablePath = new ObjectPath(dbName, tblName);
-
- TableSchema.Builder builder = new TableSchema.Builder();
- builder.fields(new String[]{"a", "m", "s"}, new DataType[]{
- DataTypes.ARRAY(DataTypes.INT()),
- DataTypes.MAP(DataTypes.INT(), DataTypes.STRING()),
- DataTypes.ROW(DataTypes.FIELD("f1", DataTypes.INT()), DataTypes.FIELD("f2", DataTypes.STRING()))});
-
- RowTypeInfo rowTypeInfo = createHiveDestTable(dbName, tblName, builder.build(), 0);
- List<Row> toWrite = new ArrayList<>();
- Row row = new Row(rowTypeInfo.getArity());
- Object[] array = new Object[]{1, 2, 3};
- Map<Integer, String> map = new HashMap<Integer, String>() {{
- put(1, "a");
- put(2, "b");
- }};
- Row struct = new Row(2);
- struct.setField(0, 3);
- struct.setField(1, "c");
-
- row.setField(0, array);
- row.setField(1, map);
- row.setField(2, struct);
- toWrite.add(row);
-
- TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode();
- Table src = tableEnv.fromTableSource(new CollectionTableSource(toWrite, rowTypeInfo));
- tableEnv.registerTable("complexSrc", src);
-
- tableEnv.registerCatalog("hive", hiveCatalog);
- tableEnv.sqlQuery("select * from complexSrc").executeInsert("hive.`default`.dest").await();
-
- List<String> result = hiveShell.executeQuery("select * from " + tblName);
- assertEquals(1, result.size());
- assertEquals("[1,2,3]\t{1:\"a\",2:\"b\"}\t{\"f1\":3,\"f2\":\"c\"}", result.get(0));
-
- hiveCatalog.dropTable(tablePath, false);
- }
-
- @Test
- public void testWriteNestedComplexType() throws Exception {
- String dbName = "default";
- String tblName = "dest";
- ObjectPath tablePath = new ObjectPath(dbName, tblName);
-
- // nested complex types
- TableSchema.Builder builder = new TableSchema.Builder();
- // array of rows
- builder.fields(new String[]{"a"}, new DataType[]{DataTypes.ARRAY(
- DataTypes.ROW(DataTypes.FIELD("f1", DataTypes.INT()), DataTypes.FIELD("f2", DataTypes.STRING())))});
- RowTypeInfo rowTypeInfo = createHiveDestTable(dbName, tblName, builder.build(), 0);
- Row row = new Row(rowTypeInfo.getArity());
- Object[] array = new Object[3];
- row.setField(0, array);
- for (int i = 0; i < array.length; i++) {
- Row struct = new Row(2);
- struct.setField(0, 1 + i);
- struct.setField(1, String.valueOf((char) ('a' + i)));
- array[i] = struct;
- }
- List<Row> toWrite = new ArrayList<>();
- toWrite.add(row);
-
- TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode();
-
- Table src = tableEnv.fromTableSource(new CollectionTableSource(toWrite, rowTypeInfo));
- tableEnv.registerTable("nestedSrc", src);
- tableEnv.registerCatalog("hive", hiveCatalog);
- tableEnv.sqlQuery("select * from nestedSrc").executeInsert("hive.`default`.dest").await();
-
- List<String> result = hiveShell.executeQuery("select * from " + tblName);
- assertEquals(1, result.size());
- assertEquals("[{\"f1\":1,\"f2\":\"a\"},{\"f1\":2,\"f2\":\"b\"},{\"f1\":3,\"f2\":\"c\"}]", result.get(0));
- hiveCatalog.dropTable(tablePath, false);
- }
-
- @Test
- public void testWriteNullValues() throws Exception {
- TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode(SqlDialect.HIVE);
- tableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
- tableEnv.useCatalog(hiveCatalog.getName());
- tableEnv.executeSql("create database db1");
- try {
- // 17 data types
- tableEnv.executeSql("create table db1.src" +
- "(t tinyint,s smallint,i int,b bigint,f float,d double,de decimal(10,5),ts timestamp,dt date," +
- "str string,ch char(5),vch varchar(8),bl boolean,bin binary,arr array<int>,mp map<int,string>,strt struct<f1:int,f2:string>)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src")
- .addRow(new Object[]{null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null})
- .commit();
- hiveShell.execute("create table db1.dest like db1.src");
-
- tableEnv.executeSql("insert into db1.dest select * from db1.src").await();
- List<String> results = hiveShell.executeQuery("select * from db1.dest");
- assertEquals(1, results.size());
- String[] cols = results.get(0).split("\t");
- assertEquals(17, cols.length);
- assertEquals("NULL", cols[0]);
- assertEquals(1, new HashSet<>(Arrays.asList(cols)).size());
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- @Test
public void testBatchAppend() throws Exception {
TableEnvironment tEnv = HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode(SqlDialect.HIVE);
tEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
@@ -436,98 +279,4 @@ public class HiveTableSinkITCase {
expected.sort(String::compareTo);
Assert.assertEquals(expected, results);
}
-
- private RowTypeInfo createHiveDestTable(String dbName, String tblName, TableSchema tableSchema, int numPartCols) throws Exception {
- CatalogTable catalogTable = createHiveCatalogTable(tableSchema, numPartCols);
- hiveCatalog.createTable(new ObjectPath(dbName, tblName), catalogTable, false);
- return new RowTypeInfo(tableSchema.getFieldTypes(), tableSchema.getFieldNames());
- }
-
- private RowTypeInfo createHiveDestTable(String dbName, String tblName, int numPartCols) throws Exception {
- TableSchema.Builder builder = new TableSchema.Builder();
- builder.fields(new String[]{"i", "l", "d", "s"},
- new DataType[]{
- DataTypes.INT(),
- DataTypes.BIGINT(),
- DataTypes.DOUBLE(),
- DataTypes.STRING()});
- return createHiveDestTable(dbName, tblName, builder.build(), numPartCols);
- }
-
- private CatalogTable createHiveCatalogTable(TableSchema tableSchema, int numPartCols) {
- if (numPartCols == 0) {
- return new CatalogTableImpl(
- tableSchema,
- new HashMap<String, String>() {{
- // creating a hive table needs explicit is_generic=false flag
- put(CatalogConfig.IS_GENERIC, String.valueOf(false));
- }},
- "");
- }
- String[] partCols = new String[numPartCols];
- System.arraycopy(tableSchema.getFieldNames(), tableSchema.getFieldNames().length - numPartCols, partCols, 0, numPartCols);
- return new CatalogTableImpl(
- tableSchema,
- Arrays.asList(partCols),
- new HashMap<String, String>() {{
- // creating a hive table needs explicit is_generic=false flag
- put(CatalogConfig.IS_GENERIC, String.valueOf(false));
- }},
- "");
- }
-
- private void verifyWrittenData(List<Row> expected, List<String> results) throws Exception {
- assertEquals(expected.size(), results.size());
- Set<String> expectedSet = new HashSet<>();
- for (int i = 0; i < results.size(); i++) {
- expectedSet.add(expected.get(i).toString().replaceAll(",", "\t"));
- }
- assertEquals(expectedSet, new HashSet<>(results));
- }
-
- private List<Row> generateRecords(int numRecords) {
- int arity = 4;
- List<Row> res = new ArrayList<>(numRecords);
- for (int i = 0; i < numRecords; i++) {
- Row row = new Row(arity);
- row.setField(0, i);
- row.setField(1, (long) i);
- row.setField(2, Double.valueOf(String.valueOf(String.format("%d.%d", i, i))));
- row.setField(3, String.valueOf((char) ('a' + i)));
- res.add(row);
- }
- return res;
- }
-
- private static class CollectionTableSource extends InputFormatTableSource<Row> {
-
- private final Collection<Row> data;
- private final RowTypeInfo rowTypeInfo;
-
- CollectionTableSource(Collection<Row> data, RowTypeInfo rowTypeInfo) {
- this.data = data;
- this.rowTypeInfo = rowTypeInfo;
- }
-
- @Override
- public DataType getProducedDataType() {
- return TypeConversions.fromLegacyInfoToDataType(rowTypeInfo).notNull();
- }
-
- @Override
- public TypeInformation<Row> getReturnType() {
- return rowTypeInfo;
- }
-
- @Override
- public InputFormat<Row, ?> getInputFormat() {
- return new CollectionInputFormat<>(data, rowTypeInfo.createSerializer(new ExecutionConfig()));
- }
-
- @Override
- public TableSchema getTableSchema() {
- return new TableSchema.Builder().fields(rowTypeInfo.getFieldNames(),
- TypeConversions.fromLegacyInfoToDataType(rowTypeInfo.getFieldTypes())).build();
- }
- }
}
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java
index 7be2fcd..9c46eb9 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java
@@ -20,7 +20,6 @@ package org.apache.flink.connectors.hive;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.configuration.ReadableConfig;
-import org.apache.flink.connectors.hive.read.HiveTableInputFormat;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -53,8 +52,6 @@ import org.apache.flink.util.CloseableIterator;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.FileUtils;
-import com.klarna.hiverunner.HiveShell;
-import com.klarna.hiverunner.annotations.HiveSQL;
import org.apache.calcite.rel.RelNode;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.mapred.JobConf;
@@ -64,7 +61,6 @@ import org.junit.Assume;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
-import org.junit.runner.RunWith;
import javax.annotation.Nullable;
@@ -94,20 +90,16 @@ import static org.mockito.Mockito.spy;
/**
* Tests {@link HiveTableSource}.
*/
-@RunWith(FlinkStandaloneHiveRunner.class)
public class HiveTableSourceITCase extends BatchAbstractTestBase {
- @HiveSQL(files = {})
- private static HiveShell hiveShell;
-
private static HiveCatalog hiveCatalog;
- private static HiveConf hiveConf;
+ private static TableEnvironment batchTableEnv;
@BeforeClass
public static void createCatalog() {
- hiveConf = hiveShell.getHiveConf();
- hiveCatalog = HiveTestUtils.createHiveCatalog(hiveConf);
+ hiveCatalog = HiveTestUtils.createHiveCatalog();
hiveCatalog.open();
+ batchTableEnv = createTableEnv();
}
@AfterClass
@@ -119,23 +111,22 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
@Before
public void setupSourceDatabaseAndData() {
- hiveShell.execute("CREATE DATABASE IF NOT EXISTS source_db");
+ batchTableEnv.executeSql("CREATE DATABASE IF NOT EXISTS source_db");
}
@Test
- public void testReadNonPartitionedTable() {
+ public void testReadNonPartitionedTable() throws Exception {
final String dbName = "source_db";
final String tblName = "test";
- TableEnvironment tEnv = createTableEnv();
- tEnv.executeSql("CREATE TABLE source_db.test ( a INT, b INT, c STRING, d BIGINT, e DOUBLE)");
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ batchTableEnv.executeSql("CREATE TABLE source_db.test ( a INT, b INT, c STRING, d BIGINT, e DOUBLE)");
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[] { 1, 1, "a", 1000L, 1.11 })
.addRow(new Object[] { 2, 2, "b", 2000L, 2.22 })
.addRow(new Object[] { 3, 3, "c", 3000L, 3.33 })
.addRow(new Object[] { 4, 4, "d", 4000L, 4.44 })
.commit();
- Table src = tEnv.sqlQuery("select * from hive.source_db.test");
+ Table src = batchTableEnv.sqlQuery("select * from hive.source_db.test");
List<Row> rows = CollectionUtil.iteratorToList(src.execute().collect());
Assert.assertEquals(4, rows.size());
@@ -149,18 +140,17 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
public void testReadComplexDataType() throws Exception {
final String dbName = "source_db";
final String tblName = "complex_test";
- TableEnvironment tEnv = createTableEnv();
- tEnv.executeSql("create table source_db.complex_test(" +
+ batchTableEnv.executeSql("create table source_db.complex_test(" +
"a array<int>, m map<int,string>, s struct<f1:int,f2:bigint>)");
Integer[] array = new Integer[]{1, 2, 3};
Map<Integer, String> map = new LinkedHashMap<>();
map.put(1, "a");
map.put(2, "b");
Object[] struct = new Object[]{3, 3L};
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{array, map, struct})
.commit();
- Table src = tEnv.sqlQuery("select * from hive.source_db.complex_test");
+ Table src = batchTableEnv.sqlQuery("select * from hive.source_db.complex_test");
List<Row> rows = CollectionUtil.iteratorToList(src.execute().collect());
Assert.assertEquals(1, rows.size());
assertArrayEquals(array, (Integer[]) rows.get(0).getField(0));
@@ -176,18 +166,17 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
public void testReadPartitionTable() throws Exception {
final String dbName = "source_db";
final String tblName = "test_table_pt";
- TableEnvironment tEnv = createTableEnv();
- tEnv.executeSql("CREATE TABLE source_db.test_table_pt " +
+ batchTableEnv.executeSql("CREATE TABLE source_db.test_table_pt " +
"(`year` STRING, `value` INT) partitioned by (pt int)");
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{"2014", 3})
.addRow(new Object[]{"2014", 4})
.commit("pt=0");
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{"2015", 2})
.addRow(new Object[]{"2015", 5})
.commit("pt=1");
- Table src = tEnv.sqlQuery("select * from hive.source_db.test_table_pt");
+ Table src = batchTableEnv.sqlQuery("select * from hive.source_db.test_table_pt");
List<Row> rows = CollectionUtil.iteratorToList(src.execute().collect());
assertEquals(4, rows.size());
@@ -199,18 +188,17 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
public void testPartitionPrunning() throws Exception {
final String dbName = "source_db";
final String tblName = "test_table_pt_1";
- TableEnvironment tEnv = createTableEnv();
- tEnv.executeSql("CREATE TABLE source_db.test_table_pt_1 " +
+ batchTableEnv.executeSql("CREATE TABLE source_db.test_table_pt_1 " +
"(`year` STRING, `value` INT) partitioned by (pt int)");
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{"2014", 3})
.addRow(new Object[]{"2014", 4})
.commit("pt=0");
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{"2015", 2})
.addRow(new Object[]{"2015", 5})
.commit("pt=1");
- Table src = tEnv.sqlQuery("select * from hive.source_db.test_table_pt_1 where pt = 0");
+ Table src = batchTableEnv.sqlQuery("select * from hive.source_db.test_table_pt_1 where pt = 0");
// first check execution plan to ensure partition prunning works
String[] explain = src.explain().split("==.*==\n");
assertEquals(4, explain.length);
@@ -225,7 +213,7 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
}
@Test
- public void testPartitionFilter() {
+ public void testPartitionFilter() throws Exception {
TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode(SqlDialect.HIVE);
TestPartitionFilterCatalog catalog = new TestPartitionFilterCatalog(
hiveCatalog.getName(), hiveCatalog.getDefaultDatabase(), hiveCatalog.getHiveConf(), hiveCatalog.getHiveVersion());
@@ -234,14 +222,14 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
tableEnv.executeSql("create database db1");
try {
tableEnv.executeSql("create table db1.part(x int) partitioned by (p1 int,p2 string)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "part")
.addRow(new Object[]{1}).commit("p1=1,p2='a'");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "part")
.addRow(new Object[]{2}).commit("p1=2,p2='b'");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "part")
.addRow(new Object[]{3}).commit("p1=3,p2='c'");
// test string partition columns with special characters
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "part")
.addRow(new Object[]{4}).commit("p1=4,p2='c:2'");
Table query = tableEnv.sqlQuery("select x from db1.part where p1>1 or p2<>'a' order by x");
String[] explain = query.explain().split("==.*==\n");
@@ -300,7 +288,7 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
}
@Test
- public void testPartitionFilterDateTimestamp() {
+ public void testPartitionFilterDateTimestamp() throws Exception {
TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode(SqlDialect.HIVE);
TestPartitionFilterCatalog catalog = new TestPartitionFilterCatalog(
hiveCatalog.getName(), hiveCatalog.getDefaultDatabase(), hiveCatalog.getHiveConf(), hiveCatalog.getHiveVersion());
@@ -309,20 +297,20 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
tableEnv.executeSql("create database db1");
try {
tableEnv.executeSql("create table db1.part(x int) partitioned by (p1 date,p2 timestamp)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part")
- .addRow(new Object[]{1}).commit("p1='2018-08-08',p2='2018-08-08 08:08:08'");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part")
- .addRow(new Object[]{2}).commit("p1='2018-08-09',p2='2018-08-08 08:08:09'");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part")
- .addRow(new Object[]{3}).commit("p1='2018-08-10',p2='2018-08-08 08:08:10'");
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "part")
+ .addRow(new Object[]{1}).commit("p1='2018-08-08',p2='2018-08-08 08:08:08.1'");
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "part")
+ .addRow(new Object[]{2}).commit("p1='2018-08-09',p2='2018-08-08 08:08:09.1'");
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "part")
+ .addRow(new Object[]{3}).commit("p1='2018-08-10',p2='2018-08-08 08:08:10.1'");
Table query = tableEnv.sqlQuery(
- "select x from db1.part where p1>cast('2018-08-09' as date) and p2<>cast('2018-08-08 08:08:09' as timestamp)");
+ "select x from db1.part where p1>cast('2018-08-09' as date) and p2<>cast('2018-08-08 08:08:09.1' as timestamp)");
String[] explain = query.explain().split("==.*==\n");
assertTrue(catalog.fallback);
String optimizedPlan = explain[2];
assertTrue(optimizedPlan, optimizedPlan.contains(
- "table=[[test-catalog, db1, part, partitions=[{p1=2018-08-10, p2=2018-08-08 08:08:10.0}]"));
+ "table=[[test-catalog, db1, part, partitions=[{p1=2018-08-10, p2=2018-08-08 08:08:10.1}]"));
List<Row> results = CollectionUtil.iteratorToList(query.execute().collect());
assertEquals("[3]", results.toString());
@@ -336,18 +324,17 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
}
@Test
- public void testProjectionPushDown() {
- TableEnvironment tableEnv = createTableEnv();
- tableEnv.executeSql("create table src(x int,y string) partitioned by (p1 bigint, p2 string)");
+ public void testProjectionPushDown() throws Exception {
+ batchTableEnv.executeSql("create table src(x int,y string) partitioned by (p1 bigint, p2 string)");
try {
- HiveTestUtils.createTextTableInserter(hiveShell, "default", "src")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "default", "src")
.addRow(new Object[]{1, "a"})
.addRow(new Object[]{2, "b"})
.commit("p1=2013, p2='2013'");
- HiveTestUtils.createTextTableInserter(hiveShell, "default", "src")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "default", "src")
.addRow(new Object[]{3, "c"})
.commit("p1=2014, p2='2014'");
- Table table = tableEnv.sqlQuery("select p1, count(y) from hive.`default`.src group by p1");
+ Table table = batchTableEnv.sqlQuery("select p1, count(y) from hive.`default`.src group by p1");
String[] explain = table.explain().split("==.*==\n");
assertEquals(4, explain.length);
String logicalPlan = explain[2];
@@ -360,24 +347,21 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
Object[] rowStrings = rows.stream().map(Row::toString).sorted().toArray();
assertArrayEquals(new String[]{"2013,2", "2014,1"}, rowStrings);
} finally {
- tableEnv.executeSql("drop table src");
+ batchTableEnv.executeSql("drop table src");
}
}
@Test
- public void testLimitPushDown() {
- TableEnvironment tableEnv = createTableEnv();
- tableEnv.executeSql("create table src (a string)");
+ public void testLimitPushDown() throws Exception {
+ batchTableEnv.executeSql("create table src (a string)");
try {
- HiveTestUtils.createTextTableInserter(hiveShell, "default", "src")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "default", "src")
.addRow(new Object[]{"a"})
.addRow(new Object[]{"b"})
.addRow(new Object[]{"c"})
.addRow(new Object[]{"d"})
.commit();
- //Add this to obtain correct stats of table to avoid FLINK-14965 problem
- hiveShell.execute("analyze table src COMPUTE STATISTICS");
- Table table = tableEnv.sqlQuery("select * from hive.`default`.src limit 1");
+ Table table = batchTableEnv.sqlQuery("select * from hive.`default`.src limit 1");
String[] explain = table.explain().split("==.*==\n");
assertEquals(4, explain.length);
String logicalPlan = explain[2];
@@ -388,29 +372,28 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
Object[] rowStrings = rows.stream().map(Row::toString).sorted().toArray();
assertArrayEquals(new String[]{"a"}, rowStrings);
} finally {
- tableEnv.executeSql("drop table src");
+ batchTableEnv.executeSql("drop table src");
}
}
@Test
- public void testParallelismSetting() {
+ public void testParallelismSetting() throws Exception {
final String dbName = "source_db";
final String tblName = "test_parallelism";
- TableEnvironment tEnv = createTableEnv();
- tEnv.executeSql("CREATE TABLE source_db.test_parallelism " +
+ batchTableEnv.executeSql("CREATE TABLE source_db.test_parallelism " +
"(`year` STRING, `value` INT) partitioned by (pt int)");
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{"2014", 3})
.addRow(new Object[]{"2014", 4})
.commit("pt=0");
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{"2015", 2})
.addRow(new Object[]{"2015", 5})
.commit("pt=1");
- Table table = tEnv.sqlQuery("select * from hive.source_db.test_parallelism");
- testParallelismSettingTranslateAndAssert(2, table, tEnv);
+ Table table = batchTableEnv.sqlQuery("select * from hive.source_db.test_parallelism");
+ testParallelismSettingTranslateAndAssert(2, table, batchTableEnv);
}
@Test
@@ -446,7 +429,7 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
}
@Test
- public void testParallelismOnLimitPushDown() {
+ public void testParallelismOnLimitPushDown() throws Exception {
final String dbName = "source_db";
final String tblName = "test_parallelism_limit_pushdown";
TableEnvironment tEnv = createTableEnv();
@@ -456,11 +439,11 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 2);
tEnv.executeSql("CREATE TABLE source_db.test_parallelism_limit_pushdown " +
"(`year` STRING, `value` INT) partitioned by (pt int)");
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{"2014", 3})
.addRow(new Object[]{"2014", 4})
.commit("pt=0");
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{"2015", 2})
.addRow(new Object[]{"2015", 5})
.commit("pt=1");
@@ -479,15 +462,15 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
// vector reader not available for 1.x and we're not testing orc for 2.0.x
Assume.assumeTrue(HiveVersionTestUtil.HIVE_210_OR_LATER);
Map<String, String> env = System.getenv();
- hiveShell.execute("create database db1");
+ batchTableEnv.executeSql("create database db1");
try {
- hiveShell.execute("create table db1.src (x int,y string) stored as orc");
- hiveShell.execute("insert into db1.src values (1,'a'),(2,'b')");
+ batchTableEnv.executeSql("create table db1.src (x int,y string) stored as orc");
+ batchTableEnv.executeSql("insert into db1.src values (1,'a'),(2,'b')").await();
testSourceConfig(true, true);
testSourceConfig(false, false);
} finally {
TestBaseUtils.setEnv(env);
- hiveShell.execute("drop database db1 cascade");
+ batchTableEnv.executeSql("drop database db1 cascade");
}
}
@@ -509,10 +492,10 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
"'streaming-source.consume-start-offset'='pt_year=2019/pt_month=09/pt_day=02'" +
")");
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{0, "a", 11})
.commit("pt_year='2019',pt_mon='09',pt_day='01'");
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{1, "b", 12})
.commit("pt_year='2020',pt_mon='09',pt_day='03'");
@@ -529,7 +512,7 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{i, "new_add", 11 + i})
.addRow(new Object[]{i, "new_add_1", 11 + i})
.commit("pt_year='2020',pt_mon='10',pt_day='0" + i + "'");
@@ -565,7 +548,7 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
")");
// the create-time is near current timestamp and bigger than '2020-10-02 00:00:00' since the code wrote
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{0, "a", 11})
.commit("p1='A1',p2='B1',p3='C1'");
@@ -582,7 +565,7 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{i, "new_add", 11 + i})
.addRow(new Object[]{i, "new_add_1", 11 + i})
.commit("p1='A',p2='B',p3='" + i + "'");
@@ -615,7 +598,7 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
"'streaming-source.consume-order'='partition-time'" +
")");
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{0, "0"})
.commit("ts='2020-05-06 00:00:00'");
@@ -632,7 +615,7 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
- HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ HiveTestUtils.createTextTableInserter(hiveCatalog, dbName, tblName)
.addRow(new Object[]{i, String.valueOf(i)})
.addRow(new Object[]{i, i + "_copy"})
.commit("ts='2020-05-06 00:" + i + "0:00'");
@@ -693,7 +676,7 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
- hiveShell.execute("insert into table source_db." + tblName + " values (1,'a'), (2,'b')");
+ batchTableEnv.executeSql("insert into table source_db." + tblName + " values (1,'a'), (2,'b')").await();
Assert.assertEquals(
Arrays.asList(Row.of(1, "a").toString(), Row.of(2, "b").toString()),
fetchRows(iter, 2));
@@ -707,12 +690,12 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
doAnswer(invocation -> {
TableSourceFactory.Context context = invocation.getArgument(0);
+ assertEquals(fallbackMR, context.getConfiguration().get(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER));
return new TestConfigSource(
new JobConf(hiveCatalog.getHiveConf()),
context.getConfiguration(),
context.getObjectIdentifier().toObjectPath(),
context.getTable(),
- fallbackMR,
inferParallelism);
}).when(tableFactorySpy).createDynamicTableSource(any(DynamicTableFactory.Context.class));
@@ -774,7 +757,6 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
* A sub-class of HiveTableSource to test vector reader switch.
*/
private static class TestConfigSource extends HiveTableSource {
- private final boolean fallbackMR;
private final boolean inferParallelism;
TestConfigSource(
@@ -782,10 +764,8 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
ReadableConfig flinkConf,
ObjectPath tablePath,
CatalogTable catalogTable,
- boolean fallbackMR,
boolean inferParallelism) {
super(jobConf, flinkConf, tablePath, catalogTable);
- this.fallbackMR = fallbackMR;
this.inferParallelism = inferParallelism;
}
@@ -796,14 +776,6 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
assertEquals(inferParallelism ? 1 : 2, parallelism);
return dataStream;
}
-
- @Override
- HiveTableInputFormat getInputFormat(
- List<HiveTablePartition> allHivePartitions,
- boolean useMapRedReader) {
- assertEquals(useMapRedReader, fallbackMR);
- return super.getInputFormat(allHivePartitions, useMapRedReader);
- }
}
// A sub-class of HiveCatalog to test list partitions by filter.
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorITCase.java
index 8bc5eb1..8dcb8d6 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorITCase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorITCase.java
@@ -18,10 +18,8 @@
package org.apache.flink.connectors.hive;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.HiveVersionTestUtil;
import org.apache.flink.table.api.SqlDialect;
-import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.constraints.UniqueConstraint;
@@ -34,40 +32,22 @@ import org.apache.flink.table.catalog.hive.client.HiveMetastoreClientFactory;
import org.apache.flink.table.catalog.hive.client.HiveMetastoreClientWrapper;
import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
import org.apache.flink.types.Row;
-import org.apache.flink.util.ArrayUtils;
import org.apache.flink.util.CollectionUtil;
-import com.klarna.hiverunner.HiveShell;
-import com.klarna.hiverunner.annotations.HiveRunnerSetup;
-import com.klarna.hiverunner.annotations.HiveSQL;
-import com.klarna.hiverunner.config.HiveRunnerConfig;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.api.Table;
-import org.apache.hadoop.hive.ql.lockmgr.DbTxnManager;
import org.junit.Assume;
import org.junit.BeforeClass;
import org.junit.Test;
-import org.junit.runner.RunWith;
-import java.sql.Date;
import java.sql.Timestamp;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_IN_TEST;
-import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY;
-import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_TXN_MANAGER;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -75,32 +55,16 @@ import static org.junit.Assert.assertTrue;
/**
* Test hive connector with table API.
*/
-@RunWith(FlinkStandaloneHiveRunner.class)
public class TableEnvHiveConnectorITCase {
- @HiveSQL(files = {})
- private static HiveShell hiveShell;
-
- @HiveRunnerSetup
- private static final HiveRunnerConfig CONFIG = new HiveRunnerConfig() {{
- if (HiveShimLoader.getHiveVersion().startsWith("3.")) {
- // hive-3.x requires a proper txn manager to create ACID table
- getHiveConfSystemOverride().put(HIVE_TXN_MANAGER.varname, DbTxnManager.class.getName());
- getHiveConfSystemOverride().put(HIVE_SUPPORT_CONCURRENCY.varname, "true");
- // tell TxnHandler to prepare txn DB
- getHiveConfSystemOverride().put(HIVE_IN_TEST.varname, "true");
- }
- }};
-
private static HiveCatalog hiveCatalog;
private static HiveMetastoreClientWrapper hmsClient;
@BeforeClass
public static void setup() {
- HiveConf hiveConf = hiveShell.getHiveConf();
- hiveCatalog = HiveTestUtils.createHiveCatalog(hiveConf);
+ hiveCatalog = HiveTestUtils.createHiveCatalog();
hiveCatalog.open();
- hmsClient = HiveMetastoreClientFactory.create(hiveConf, HiveShimLoader.getHiveVersion());
+ hmsClient = HiveMetastoreClientFactory.create(hiveCatalog.getHiveConf(), HiveShimLoader.getHiveVersion());
}
@Test
@@ -109,11 +73,11 @@ public class TableEnvHiveConnectorITCase {
tableEnv.executeSql("create database db1");
tableEnv.executeSql("create table db1.src (x int, y int)");
tableEnv.executeSql("create table db1.part (x int) partitioned by (y int)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src").addRow(new Object[]{1, 1}).addRow(new Object[]{2, null}).commit();
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "src").addRow(new Object[]{1, 1}).addRow(new Object[]{2, null}).commit();
// test generating partitions with default name
tableEnv.executeSql("insert into db1.part select * from db1.src").await();
- HiveConf hiveConf = hiveShell.getHiveConf();
+ HiveConf hiveConf = hiveCatalog.getHiveConf();
String defaultPartName = hiveConf.getVar(HiveConf.ConfVars.DEFAULTPARTITIONNAME);
Table hiveTable = hmsClient.getTable("db1", "part");
Path defaultPartPath = new Path(hiveTable.getSd().getLocation(), "y=" + defaultPartName);
@@ -141,209 +105,16 @@ public class TableEnvHiveConnectorITCase {
}
@Test
- public void testDifferentFormats() throws Exception {
- String[] formats = new String[]{"orc", "parquet", "sequencefile", "csv", "avro"};
- for (String format : formats) {
- if (format.equals("avro") && !HiveVersionTestUtil.HIVE_110_OR_LATER) {
- // timestamp is not supported for avro tables before 1.1.0
- continue;
- }
- readWriteFormat(format);
- }
- }
-
- private void readWriteFormat(String format) throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
-
- tableEnv.executeSql("create database db1");
-
- // create source and dest tables
- String suffix;
- if (format.equals("csv")) {
- suffix = "row format serde 'org.apache.hadoop.hive.serde2.OpenCSVSerde'";
- } else {
- suffix = "stored as " + format;
- }
- String tableSchema;
- // use 2018-08-20 00:00:00.1 to avoid multi-version print difference.
- List<Object> row1 = new ArrayList<>(Arrays.asList(1, "a", "2018-08-20 00:00:00.1"));
- List<Object> row2 = new ArrayList<>(Arrays.asList(2, "b", "2019-08-26 00:00:00.1"));
- // some data types are not supported for parquet tables in early versions -- https://issues.apache.org/jira/browse/HIVE-6384
- if (HiveVersionTestUtil.HIVE_120_OR_LATER || !format.equals("parquet")) {
- tableSchema = "(i int,s string,ts timestamp,dt date)";
- row1.add("2018-08-20");
- row2.add("2019-08-26");
- } else {
- tableSchema = "(i int,s string,ts timestamp)";
- }
-
- tableEnv.executeSql(String.format(
- "create table db1.src %s partitioned by (p1 string, p2 timestamp) %s", tableSchema, suffix));
- tableEnv.executeSql(String.format(
- "create table db1.dest %s partitioned by (p1 string, p2 timestamp) %s", tableSchema, suffix));
-
- // prepare source data with Hive
- // TABLE keyword in INSERT INTO is mandatory prior to 1.1.0
- hiveShell.execute(String.format(
- "insert into table db1.src partition(p1='first',p2='2018-08-20 00:00:00.1') values (%s)",
- toRowValue(row1)));
- hiveShell.execute(String.format(
- "insert into table db1.src partition(p1='second',p2='2018-08-26 00:00:00.1') values (%s)",
- toRowValue(row2)));
-
- List<String> expected = Arrays.asList(
- String.join("\t", ArrayUtils.concat(
- row1.stream().map(Object::toString).toArray(String[]::new),
- new String[]{"first", "2018-08-20 00:00:00.1"})),
- String.join("\t", ArrayUtils.concat(
- row2.stream().map(Object::toString).toArray(String[]::new),
- new String[]{"second", "2018-08-26 00:00:00.1"})));
-
- verifyFlinkQueryResult(tableEnv.sqlQuery("select * from db1.src"), expected);
-
- // Ignore orc write test for Hive version 2.0.x for now due to FLINK-13998
- if (!format.equals("orc") || !HiveShimLoader.getHiveVersion().startsWith("2.0")) {
- // populate dest table with source table
- tableEnv.executeSql("insert into db1.dest select * from db1.src").await();
-
- // verify data on hive side
- verifyHiveQueryResult("select * from db1.dest", expected);
- }
-
- tableEnv.executeSql("drop database db1 cascade");
- }
-
- private String toRowValue(List<Object> row) {
- return row.stream().map(o -> {
- String res = o.toString();
- if (o instanceof String) {
- res = "'" + res + "'";
- }
- return res;
- }).collect(Collectors.joining(","));
- }
-
- @Test
- public void testDecimal() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.src1 (x decimal(10,2))");
- tableEnv.executeSql("create table db1.src2 (x decimal(10,2))");
- tableEnv.executeSql("create table db1.dest (x decimal(10,2))");
- // populate src1 from Hive
- // TABLE keyword in INSERT INTO is mandatory prior to 1.1.0
- hiveShell.execute("insert into table db1.src1 values (1.0),(2.12),(5.123),(5.456),(123456789.12)");
-
- // populate src2 with same data from Flink
- tableEnv.executeSql("insert into db1.src2 values (cast(1.0 as decimal(10,2))), (cast(2.12 as decimal(10,2))), " +
- "(cast(5.123 as decimal(10,2))), (cast(5.456 as decimal(10,2))), (cast(123456789.12 as decimal(10,2)))")
- .await();
- // verify src1 and src2 contain same data
- verifyHiveQueryResult("select * from db1.src2", hiveShell.executeQuery("select * from db1.src1"));
-
- // populate dest with src1 from Flink -- to test reading decimal type from Hive
- tableEnv.executeSql("insert into db1.dest select * from db1.src1").await();
- verifyHiveQueryResult("select * from db1.dest", hiveShell.executeQuery("select * from db1.src1"));
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- @Test
- public void testInsertOverwrite() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- // non-partitioned
- tableEnv.executeSql("create table db1.dest (x int, y string)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "dest").addRow(new Object[]{1, "a"}).addRow(new Object[]{2, "b"}).commit();
- verifyHiveQueryResult("select * from db1.dest", Arrays.asList("1\ta", "2\tb"));
-
- tableEnv.executeSql("insert overwrite db1.dest values (3, 'c')").await();
- verifyHiveQueryResult("select * from db1.dest", Collections.singletonList("3\tc"));
-
- // static partition
- tableEnv.executeSql("create table db1.part(x int) partitioned by (y int)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part").addRow(new Object[]{1}).commit("y=1");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part").addRow(new Object[]{2}).commit("y=2");
- tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("insert overwrite db1.part partition (y=1) select 100").await();
- verifyHiveQueryResult("select * from db1.part", Arrays.asList("100\t1", "2\t2"));
-
- // dynamic partition
- tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("insert overwrite db1.part values (200,2),(3,3)").await();
- // only overwrite dynamically matched partitions, other existing partitions remain intact
- verifyHiveQueryResult("select * from db1.part", Arrays.asList("100\t1", "200\t2", "3\t3"));
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- @Test
- public void testStaticPartition() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.src (x int)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src").addRow(new Object[]{1}).addRow(new Object[]{2}).commit();
- tableEnv.executeSql("create table db1.dest (x int) partitioned by (p1 string, p2 double)");
- tableEnv.executeSql("insert into db1.dest partition (p1='1''1', p2=1.1) select x from db1.src").await();
- assertEquals(1, hiveCatalog.listPartitions(new ObjectPath("db1", "dest")).size());
- verifyHiveQueryResult("select * from db1.dest", Arrays.asList("1\t1'1\t1.1", "2\t1'1\t1.1"));
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- @Test
- public void testDynamicPartition() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.src (x int, y string, z double)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src")
- .addRow(new Object[]{1, "a", 1.1})
- .addRow(new Object[]{2, "a", 2.2})
- .addRow(new Object[]{3, "b", 3.3})
- .commit();
- tableEnv.executeSql("create table db1.dest (x int) partitioned by (p1 string, p2 double)");
- tableEnv.executeSql("insert into db1.dest select * from db1.src").await();
- assertEquals(3, hiveCatalog.listPartitions(new ObjectPath("db1", "dest")).size());
- verifyHiveQueryResult("select * from db1.dest", Arrays.asList("1\ta\t1.1", "2\ta\t2.2", "3\tb\t3.3"));
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- @Test
- public void testPartialDynamicPartition() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.src (x int, y string)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src").addRow(new Object[]{1, "a"}).addRow(new Object[]{2, "b"}).commit();
- tableEnv.executeSql("create table db1.dest (x int) partitioned by (p1 double, p2 string)");
- tableEnv.executeSql("insert into db1.dest partition (p1=1.1) select x,y from db1.src").await();
- assertEquals(2, hiveCatalog.listPartitions(new ObjectPath("db1", "dest")).size());
- verifyHiveQueryResult("select * from db1.dest", Arrays.asList("1\t1.1\ta", "2\t1.1\tb"));
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- @Test
public void testDateTimestampPartitionColumns() throws Exception {
TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
tableEnv.executeSql("create database db1");
try {
tableEnv.executeSql("create table db1.part(x int) partitioned by (dt date,ts timestamp)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "part")
.addRow(new Object[]{1})
.addRow(new Object[]{2})
.commit("dt='2019-12-23',ts='2019-12-23 00:00:00'");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "part")
.addRow(new Object[]{3})
.commit("dt='2019-12-25',ts='2019-12-25 16:23:43.012'");
List<Row> results = CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.part order by x").execute().collect());
@@ -362,7 +133,7 @@ public class TableEnvHiveConnectorITCase {
}
@Test
- public void testUDTF() {
+ public void testUDTF() throws Exception {
// W/o https://issues.apache.org/jira/browse/HIVE-11878 Hive registers the App classloader as the classloader
// for the UDTF and closes the App classloader when we tear down the session. This causes problems for JUnit code
// and shutdown hooks that have to run after the test finishes, because App classloader can no longer load new
@@ -376,13 +147,17 @@ public class TableEnvHiveConnectorITCase {
tableEnv.executeSql("create table db1.simple (i int,a array<int>)");
tableEnv.executeSql("create table db1.nested (a array<map<int, string>>)");
tableEnv.executeSql("create function hiveudtf as 'org.apache.hadoop.hive.ql.udf.generic.GenericUDTFExplode'");
- hiveShell.insertInto("db1", "simple").addRow(3, Arrays.asList(1, 2, 3)).commit();
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "simple")
+ .addRow(new Object[]{3, Arrays.asList(1, 2, 3)})
+ .commit();
Map<Integer, String> map1 = new HashMap<>();
map1.put(1, "a");
map1.put(2, "b");
Map<Integer, String> map2 = new HashMap<>();
map2.put(3, "c");
- hiveShell.insertInto("db1", "nested").addRow(Arrays.asList(map1, map2)).commit();
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "nested")
+ .addRow(new Object[]{Arrays.asList(map1, map2)})
+ .commit();
List<Row> results = CollectionUtil.iteratorToList(
tableEnv.sqlQuery("select x from db1.simple, lateral table(hiveudtf(a)) as T(x)").execute().collect());
@@ -392,7 +167,7 @@ public class TableEnvHiveConnectorITCase {
assertEquals("[{1=a, 2=b}, {3=c}]", results.toString());
tableEnv.executeSql("create table db1.ts (a array<timestamp>)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "ts").addRow(new Object[]{
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "ts").addRow(new Object[]{
new Object[]{Timestamp.valueOf("2015-04-28 15:23:00"), Timestamp.valueOf("2016-06-03 17:05:52")}})
.commit();
results = CollectionUtil.iteratorToList(
@@ -458,176 +233,6 @@ public class TableEnvHiveConnectorITCase {
}
@Test
- public void testTimestamp() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.src (ts timestamp)");
- tableEnv.executeSql("create table db1.dest (ts timestamp)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src")
- .addRow(new Object[]{Timestamp.valueOf("2019-11-11 00:00:00")})
- .addRow(new Object[]{Timestamp.valueOf("2019-12-03 15:43:32.123456789")})
- .commit();
- // test read timestamp from hive
- List<Row> results = CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.src").execute().collect());
- assertEquals(2, results.size());
- assertEquals(LocalDateTime.of(2019, 11, 11, 0, 0), results.get(0).getField(0));
- assertEquals(LocalDateTime.of(2019, 12, 3, 15, 43, 32, 123456789), results.get(1).getField(0));
- // test write timestamp to hive
- tableEnv.executeSql("insert into db1.dest select max(ts) from db1.src").await();
- verifyHiveQueryResult("select * from db1.dest", Collections.singletonList("2019-12-03 15:43:32.123456789"));
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- @Test
- public void testDate() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.src (dt date)");
- tableEnv.executeSql("create table db1.dest (dt date)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src")
- .addRow(new Object[]{Date.valueOf("2019-12-09")})
- .addRow(new Object[]{Date.valueOf("2019-12-12")})
- .commit();
- // test read date from hive
- List<Row> results = CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.src").execute().collect());
- assertEquals(2, results.size());
- assertEquals(LocalDate.of(2019, 12, 9), results.get(0).getField(0));
- assertEquals(LocalDate.of(2019, 12, 12), results.get(1).getField(0));
- // test write date to hive
- tableEnv.executeSql("insert into db1.dest select max(dt) from db1.src").await();
- verifyHiveQueryResult("select * from db1.dest", Collections.singletonList("2019-12-12"));
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- @Test
- public void testViews() {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.src (key int,val string)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src")
- .addRow(new Object[]{1, "a"})
- .addRow(new Object[]{1, "aa"})
- .addRow(new Object[]{1, "aaa"})
- .addRow(new Object[]{2, "b"})
- .addRow(new Object[]{3, "c"})
- .addRow(new Object[]{3, "ccc"})
- .commit();
- tableEnv.executeSql("create table db1.keys (key int,name string)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "keys")
- .addRow(new Object[]{1, "key1"})
- .addRow(new Object[]{2, "key2"})
- .addRow(new Object[]{3, "key3"})
- .addRow(new Object[]{4, "key4"})
- .commit();
- hiveShell.execute("create view db1.v1 as select key as k,val as v from db1.src limit 2");
- hiveShell.execute("create view db1.v2 as select key,count(*) from db1.src group by key having count(*)>1 order by key");
- hiveShell.execute("create view db1.v3 as select k.key,k.name,count(*) from db1.src s join db1.keys k on s.key=k.key group by k.key,k.name order by k.key");
- List<Row> results = CollectionUtil.iteratorToList(tableEnv.sqlQuery("select count(v) from db1.v1").execute().collect());
- assertEquals("[2]", results.toString());
- results = CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.v2").execute().collect());
- assertEquals("[1,3, 3,2]", results.toString());
- results = CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.v3").execute().collect());
- assertEquals("[1,key1,3, 2,key2,1, 3,key3,2]", results.toString());
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- @Test
- public void testWhitespacePartValue() throws Exception {
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.dest (x int) partitioned by (p string)");
- StatementSet stmtSet = tableEnv.createStatementSet();
- stmtSet.addInsertSql("insert into db1.dest select 1,' '");
- stmtSet.addInsertSql("insert into db1.dest select 2,'a \t'");
- stmtSet.execute().await();
- assertEquals("[p= , p=a %09]", hiveShell.executeQuery("show partitions db1.dest").toString());
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- private void testCompressTextTable(boolean batch) throws Exception {
- TableEnvironment tableEnv = batch ?
- getTableEnvWithHiveCatalog() :
- getStreamTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.src (x string,y string)");
- hiveShell.execute("create table db1.dest like db1.src");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src")
- .addRow(new Object[]{"a", "b"})
- .addRow(new Object[]{"c", "d"})
- .commit();
- hiveCatalog.getHiveConf().setBoolVar(HiveConf.ConfVars.COMPRESSRESULT, true);
- tableEnv.executeSql("insert into db1.dest select * from db1.src").await();
- List<String> expected = Arrays.asList("a\tb", "c\td");
- verifyHiveQueryResult("select * from db1.dest", expected);
- verifyFlinkQueryResult(tableEnv.sqlQuery("select * from db1.dest"), expected);
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- @Test
- public void testBatchCompressTextTable() throws Exception {
- testCompressTextTable(true);
- }
-
- @Test
- public void testStreamCompressTextTable() throws Exception {
- testCompressTextTable(false);
- }
-
- private void testTransactionalTable(boolean batch) {
- TableEnvironment tableEnv = batch ?
- getTableEnvWithHiveCatalog() :
- getStreamTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.src (x string,y string)");
- hiveShell.execute("create table db1.dest (x string,y string) clustered by (x) into 3 buckets stored as orc tblproperties ('transactional'='true')");
- List<Exception> exceptions = new ArrayList<>();
- try {
- tableEnv.executeSql("insert into db1.src select * from db1.dest").await();
- } catch (Exception e) {
- exceptions.add(e);
- }
- try {
- tableEnv.executeSql("insert into db1.dest select * from db1.src").await();
- } catch (Exception e) {
- exceptions.add(e);
- }
- assertEquals(2, exceptions.size());
- exceptions.forEach(e -> {
- assertTrue(e instanceof FlinkHiveException);
- assertEquals("Reading or writing ACID table db1.dest is not supported.", e.getMessage());
- });
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- @Test
- public void testBatchTransactionalTable() {
- testTransactionalTable(true);
- }
-
- @Test
- public void testStreamTransactionalTable() {
- testTransactionalTable(false);
- }
-
- @Test
public void testRegexSerDe() throws Exception {
TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
tableEnv.executeSql("create database db1");
@@ -635,7 +240,7 @@ public class TableEnvHiveConnectorITCase {
tableEnv.executeSql("create table db1.src (x int,y string) " +
"row format serde 'org.apache.hadoop.hive.serde2.RegexSerDe' " +
"with serdeproperties ('input.regex'='([\\\\d]+)\\u0001([\\\\S]+)')");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "src")
.addRow(new Object[]{1, "a"})
.addRow(new Object[]{2, "ab"})
.commit();
@@ -679,40 +284,17 @@ public class TableEnvHiveConnectorITCase {
}
@Test
- public void testOrcSchemaEvol() throws Exception {
- // not supported until 2.1.0 -- https://issues.apache.org/jira/browse/HIVE-11981,
- // https://issues.apache.org/jira/browse/HIVE-13178
- Assume.assumeTrue(HiveVersionTestUtil.HIVE_210_OR_LATER);
- TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
- tableEnv.executeSql("create database db1");
- try {
- tableEnv.executeSql("create table db1.src (x smallint,y int) stored as orc");
- hiveShell.execute("insert into table db1.src values (1,100),(2,200)");
-
- tableEnv.getConfig().getConfiguration().setBoolean(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER, true);
-
- tableEnv.executeSql("alter table db1.src change x x int");
- assertEquals("[1,100, 2,200]", CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.src").execute().collect()).toString());
-
- tableEnv.executeSql("alter table db1.src change y y string");
- assertEquals("[1,100, 2,200]", CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.src").execute().collect()).toString());
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
- }
-
- @Test
public void testNonExistingPartitionFolder() throws Exception {
TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
tableEnv.executeSql("create database db1");
try {
tableEnv.executeSql("create table db1.part (x int) partitioned by (p int)");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part").addRow(new Object[]{1}).commit("p=1");
- HiveTestUtils.createTextTableInserter(hiveShell, "db1", "part").addRow(new Object[]{2}).commit("p=2");
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "part").addRow(new Object[]{1}).commit("p=1");
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "db1", "part").addRow(new Object[]{2}).commit("p=2");
tableEnv.executeSql("alter table db1.part add partition (p=3)");
// remove one partition
Path toRemove = new Path(hiveCatalog.getHiveTable(new ObjectPath("db1", "part")).getSd().getLocation(), "p=2");
- FileSystem fs = toRemove.getFileSystem(hiveShell.getHiveConf());
+ FileSystem fs = toRemove.getFileSystem(hiveCatalog.getHiveConf());
fs.delete(toRemove, true);
List<Row> results = CollectionUtil.iteratorToList(tableEnv.sqlQuery("select * from db1.part").execute().collect());
@@ -727,7 +309,7 @@ public class TableEnvHiveConnectorITCase {
TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
tableEnv.executeSql("create table src (x int,y string)");
HiveTestUtils.createTextTableInserter(
- hiveShell,
+ hiveCatalog,
"default",
"src")
.addRow(new Object[]{1, "a"})
@@ -756,31 +338,4 @@ public class TableEnvHiveConnectorITCase {
tableEnv.useCatalog(hiveCatalog.getName());
return tableEnv;
}
-
- private TableEnvironment getStreamTableEnvWithHiveCatalog() {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerStreamMode(env, SqlDialect.HIVE);
- tableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
- tableEnv.useCatalog(hiveCatalog.getName());
- return tableEnv;
- }
-
- private void verifyHiveQueryResult(String query, List<String> expected) {
- List<String> results = hiveShell.executeQuery(query);
- assertEquals(expected.size(), results.size());
- assertEquals(new HashSet<>(expected), new HashSet<>(results));
- }
-
- private void verifyFlinkQueryResult(org.apache.flink.table.api.Table table, List<String> expected) throws Exception {
- List<Row> rows = CollectionUtil.iteratorToList(table.execute().collect());
- List<String> results = rows.stream().map(row ->
- IntStream.range(0, row.getArity())
- .mapToObj(row::getField)
- .map(o -> o instanceof LocalDateTime ?
- Timestamp.valueOf((LocalDateTime) o) : o)
- .map(Object::toString)
- .collect(Collectors.joining("\t"))).collect(Collectors.toList());
- assertEquals(expected.size(), results.size());
- assertEquals(new HashSet<>(expected), new HashSet<>(results));
- }
}
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogITCase.java
index f7eaeec..3f9cc5d 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogITCase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogITCase.java
@@ -20,7 +20,6 @@ package org.apache.flink.table.catalog.hive;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
-import org.apache.flink.connectors.hive.FlinkStandaloneHiveRunner;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
@@ -40,16 +39,12 @@ import org.apache.flink.types.Row;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.FileUtils;
-import com.klarna.hiverunner.HiveShell;
-import com.klarna.hiverunner.annotations.HiveSQL;
-import org.apache.hadoop.hive.conf.HiveConf;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
@@ -83,25 +78,19 @@ import static org.junit.Assert.assertNull;
* IT case for HiveCatalog.
* TODO: move to flink-connector-hive-test end-to-end test module once it's setup
*/
-@RunWith(FlinkStandaloneHiveRunner.class)
public class HiveCatalogITCase {
- @HiveSQL(files = {})
- private static HiveShell hiveShell;
-
@Rule
public TemporaryFolder tempFolder = new TemporaryFolder();
private static HiveCatalog hiveCatalog;
- private static HiveConf hiveConf;
private String sourceTableName = "csv_source";
private String sinkTableName = "csv_sink";
@BeforeClass
public static void createCatalog() {
- hiveConf = hiveShell.getHiveConf();
- hiveCatalog = HiveTestUtils.createHiveCatalog(hiveConf);
+ hiveCatalog = HiveTestUtils.createHiveCatalog();
hiveCatalog.open();
}
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogUseBlinkITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogUseBlinkITCase.java
index da25307..af693ee 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogUseBlinkITCase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogUseBlinkITCase.java
@@ -20,7 +20,6 @@ package org.apache.flink.table.catalog.hive;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.connectors.hive.FlinkStandaloneHiveRunner;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.DataTypes;
@@ -48,9 +47,6 @@ import org.apache.flink.types.Row;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.FileUtils;
-import com.klarna.hiverunner.HiveShell;
-import com.klarna.hiverunner.annotations.HiveSQL;
-import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.udf.UDFMonth;
import org.apache.hadoop.hive.ql.udf.UDFYear;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFSum;
@@ -60,7 +56,6 @@ import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
import java.io.IOException;
import java.nio.file.Files;
@@ -80,12 +75,8 @@ import static java.lang.String.format;
* IT case for HiveCatalog.
* TODO: move to flink-connector-hive-test end-to-end test module once it's setup
*/
-@RunWith(FlinkStandaloneHiveRunner.class)
public class HiveCatalogUseBlinkITCase extends AbstractTestBase {
- @HiveSQL(files = {})
- private static HiveShell hiveShell;
-
@Rule
public TemporaryFolder tempFolder = new TemporaryFolder();
@@ -96,8 +87,7 @@ public class HiveCatalogUseBlinkITCase extends AbstractTestBase {
@BeforeClass
public static void createCatalog() {
- HiveConf hiveConf = hiveShell.getHiveConf();
- hiveCatalog = HiveTestUtils.createHiveCatalog(hiveConf);
+ hiveCatalog = HiveTestUtils.createHiveCatalog();
hiveCatalog.open();
}
@@ -248,7 +238,7 @@ public class HiveCatalogUseBlinkITCase extends AbstractTestBase {
}
@Test
- public void testTimestampUDF() {
+ public void testTimestampUDF() throws Exception {
TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode(SqlDialect.HIVE);
tableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
@@ -256,7 +246,7 @@ public class HiveCatalogUseBlinkITCase extends AbstractTestBase {
tableEnv.executeSql(String.format("create function myyear as '%s'", UDFYear.class.getName()));
tableEnv.executeSql("create table src(ts timestamp)");
try {
- HiveTestUtils.createTextTableInserter(hiveShell, "default", "src")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "default", "src")
.addRow(new Object[]{Timestamp.valueOf("2013-07-15 10:00:00")})
.addRow(new Object[]{Timestamp.valueOf("2019-05-23 17:32:55")})
.commit();
@@ -271,7 +261,7 @@ public class HiveCatalogUseBlinkITCase extends AbstractTestBase {
}
@Test
- public void testDateUDF() {
+ public void testDateUDF() throws Exception {
TableEnvironment tableEnv = HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode(SqlDialect.HIVE);
tableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
@@ -279,7 +269,7 @@ public class HiveCatalogUseBlinkITCase extends AbstractTestBase {
tableEnv.executeSql(String.format("create function mymonth as '%s'", UDFMonth.class.getName()));
tableEnv.executeSql("create table src(dt date)");
try {
- HiveTestUtils.createTextTableInserter(hiveShell, "default", "src")
+ HiveTestUtils.createTextTableInserter(hiveCatalog, "default", "src")
.addRow(new Object[]{Date.valueOf("2019-01-19")})
.addRow(new Object[]{Date.valueOf("2019-03-02")})
.commit();
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveTestUtils.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveTestUtils.java
index 8aa25b8..0cbdfe4 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveTestUtils.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveTestUtils.java
@@ -18,18 +18,29 @@
package org.apache.flink.table.catalog.hive;
+import org.apache.flink.sql.parser.SqlPartitionUtils;
+import org.apache.flink.sql.parser.hive.ddl.SqlAddHivePartitions;
+import org.apache.flink.sql.parser.hive.impl.FlinkHiveSqlParserImpl;
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.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.CatalogPartitionSpec;
import org.apache.flink.table.catalog.CatalogTest;
+import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
+import org.apache.flink.util.Preconditions;
import org.apache.flink.util.StringUtils;
-import com.klarna.hiverunner.HiveShell;
+import org.apache.calcite.config.Lex;
+import org.apache.calcite.sql.parser.SqlParser;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
import org.junit.rules.TemporaryFolder;
import java.io.BufferedWriter;
@@ -58,7 +69,7 @@ public class HiveTestUtils {
private static final int MIN_EPH_PORT = 49152;
private static final int MAX_EPH_PORT = 61000;
- private static final byte[] SEPARATORS = new byte[]{(byte) 1, (byte) 2, (byte) 3};
+ private static final byte[] SEPARATORS = new byte[]{(byte) 1, (byte) 2, (byte) 3, (byte) 4, (byte) 5, (byte) 6, (byte) 7, (byte) 8};
/**
* Create a HiveCatalog with an embedded Hive Metastore.
@@ -151,8 +162,8 @@ public class HiveTestUtils {
}
// Insert into a single partition of a text table.
- public static TextTableInserter createTextTableInserter(HiveShell hiveShell, String dbName, String tableName) {
- return new TextTableInserter(hiveShell, dbName, tableName);
+ public static TextTableInserter createTextTableInserter(HiveCatalog hiveCatalog, String dbName, String tableName) {
+ return new TextTableInserter(hiveCatalog, dbName, tableName);
}
/**
@@ -160,13 +171,16 @@ public class HiveTestUtils {
*/
public static class TextTableInserter {
- private final HiveShell hiveShell;
+ private final HiveCatalog hiveCatalog;
+ private final TableEnvironment tableEnv;
private final String dbName;
private final String tableName;
private final List<Object[]> rows;
- public TextTableInserter(HiveShell hiveShell, String dbName, String tableName) {
- this.hiveShell = hiveShell;
+ public TextTableInserter(HiveCatalog hiveCatalog, String dbName, String tableName) {
+ this.hiveCatalog = hiveCatalog;
+ tableEnv = createTableEnvWithHiveCatalog(hiveCatalog);
+ tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
this.dbName = dbName;
this.tableName = tableName;
rows = new ArrayList<>();
@@ -177,31 +191,42 @@ public class HiveTestUtils {
return this;
}
- public void commit() {
+ public void commit() throws Exception {
commit(null);
}
- public void commit(String partitionSpec) {
- try {
- File file = File.createTempFile("table_data_", null);
- try (BufferedWriter writer = new BufferedWriter(new FileWriter(file))) {
- for (int i = 0; i < rows.size(); i++) {
- if (i > 0) {
- writer.newLine();
- }
- writer.write(toText(rows.get(i)));
+ public void commit(String partitionSpec) throws Exception {
+ File file = File.createTempFile("table_data_", null);
+ try (BufferedWriter writer = new BufferedWriter(new FileWriter(file))) {
+ for (int i = 0; i < rows.size(); i++) {
+ if (i > 0) {
+ writer.newLine();
}
- // new line at the end of file
- writer.newLine();
+ writer.write(toText(rows.get(i)));
}
- String load = String.format("load data local inpath '%s' into table %s.%s", file.getAbsolutePath(), dbName, tableName);
- if (partitionSpec != null) {
- load += String.format(" partition (%s)", partitionSpec);
- }
- hiveShell.execute(load);
- } catch (IOException e) {
- throw new RuntimeException(e);
+ // new line at the end of file
+ writer.newLine();
+ }
+ Path src = new Path(file.toURI());
+ Path dest;
+ ObjectPath tablePath = new ObjectPath(dbName, tableName);
+ Table hiveTable = hiveCatalog.getHiveTable(tablePath);
+ if (partitionSpec != null) {
+ String ddl = String.format("alter table `%s`.`%s` add if not exists partition (%s)", dbName, tableName, partitionSpec);
+ tableEnv.executeSql(ddl);
+ // we need parser to parse the partition spec
+ SqlParser parser = SqlParser.create(
+ ddl,
+ SqlParser.config().withParserFactory(FlinkHiveSqlParserImpl.FACTORY).withLex(Lex.JAVA));
+ SqlAddHivePartitions sqlAddPart = (SqlAddHivePartitions) parser.parseStmt();
+ Map<String, String> spec = SqlPartitionUtils.getPartitionKVs(sqlAddPart.getPartSpecs().get(0));
+ Partition hivePart = hiveCatalog.getHivePartition(hiveTable, new CatalogPartitionSpec(spec));
+ dest = new Path(hivePart.getSd().getLocation(), src.getName());
+ } else {
+ dest = new Path(hiveTable.getSd().getLocation(), src.getName());
}
+ FileSystem fs = dest.getFileSystem(hiveCatalog.getHiveConf());
+ Preconditions.checkState(fs.rename(src, dest));
}
private String toText(Object[] row) {
@@ -210,7 +235,7 @@ public class HiveTestUtils {
if (builder.length() > 0) {
builder.appendCodePoint(SEPARATORS[0]);
}
- String colStr = toText(col);
+ String colStr = toText(col, 1);
if (colStr != null) {
builder.append(colStr);
}
@@ -218,7 +243,7 @@ public class HiveTestUtils {
return builder.toString();
}
- private String toText(Object obj) {
+ private String toText(Object obj, final int level) {
if (obj == null) {
return null;
}
@@ -226,26 +251,26 @@ public class HiveTestUtils {
if (obj instanceof Map) {
for (Object key : ((Map) obj).keySet()) {
if (builder.length() > 0) {
- builder.appendCodePoint(SEPARATORS[1]);
+ builder.appendCodePoint(SEPARATORS[level]);
}
- builder.append(toText(key));
- builder.appendCodePoint(SEPARATORS[2]);
- builder.append(toText(((Map) obj).get(key)));
+ builder.append(toText(key, level + 2));
+ builder.appendCodePoint(SEPARATORS[level + 1]);
+ builder.append(toText(((Map) obj).get(key), level + 2));
}
} else if (obj instanceof Object[]) {
Object[] array = (Object[]) obj;
for (Object element : array) {
if (builder.length() > 0) {
- builder.appendCodePoint(SEPARATORS[1]);
+ builder.appendCodePoint(SEPARATORS[level]);
}
- builder.append(toText(element));
+ builder.append(toText(element, level + 1));
}
} else if (obj instanceof List) {
for (Object element : (List) obj) {
if (builder.length() > 0) {
- builder.appendCodePoint(SEPARATORS[1]);
+ builder.appendCodePoint(SEPARATORS[level]);
}
- builder.append(toText(element));
+ builder.append(toText(element, level + 1));
}
} else {
builder.append(obj);