You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by bl...@apache.org on 2019/10/29 17:29:43 UTC
[flink] branch master updated: [FLINK-14526][hive] Support Hive
version 1.1.0 and 1.1.1
This is an automated email from the ASF dual-hosted git repository.
bli 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 6cc09ac [FLINK-14526][hive] Support Hive version 1.1.0 and 1.1.1
6cc09ac is described below
commit 6cc09ac6ed822e2b3d753b27c621f5122d353673
Author: Rui Li <li...@apache.org>
AuthorDate: Fri Oct 25 19:13:55 2019 +0800
[FLINK-14526][hive] Support Hive version 1.1.0 and 1.1.1
To support Hive 1.1.0 and 1.1.1.
This closes #9995.
---
flink-connectors/flink-connector-hive/pom.xml | 23 +++
.../flink/table/catalog/hive/HiveCatalog.java | 13 +-
.../flink/table/catalog/hive/client/HiveShim.java | 27 ++-
.../table/catalog/hive/client/HiveShimLoader.java | 8 +
.../{HiveShimV120.java => HiveShimV110.java} | 122 ++++++++++--
.../table/catalog/hive/client/HiveShimV111.java} | 25 +--
.../table/catalog/hive/client/HiveShimV120.java | 218 +++++++++------------
.../catalog/hive/util/HiveReflectionUtils.java | 28 +--
.../table/catalog/hive/util/HiveStatsUtil.java | 48 ++---
.../table/catalog/hive/util/HiveTypeUtil.java | 9 +
.../flink/table/functions/hive/HiveGenericUDF.java | 10 +-
.../table/functions/hive/HiveGenericUDTF.java | 2 +-
.../functions/hive/conversion/HiveInspectors.java | 125 ++++++------
.../conversion/WritableHiveObjectConversion.java} | 29 +--
.../hive/FlinkStandaloneHiveServerContext.java | 3 +
.../connectors/hive/HiveRunnerShimLoader.java | 2 +
.../flink/connectors/hive/HiveTableSourceTest.java | 58 +++---
.../connectors/hive/TableEnvHiveConnectorTest.java | 17 +-
.../catalog/hive/HiveCatalogHiveMetadataTest.java | 33 ++--
.../table/catalog/hive/HiveCatalogTestBase.java | 11 ++
.../flink/table/catalog/hive/HiveTestUtils.java | 109 +++++++++++
.../table/functions/hive/HiveGenericUDFTest.java | 10 +-
.../table/functions/hive/HiveGenericUDTFTest.java | 12 +-
23 files changed, 594 insertions(+), 348 deletions(-)
diff --git a/flink-connectors/flink-connector-hive/pom.xml b/flink-connectors/flink-connector-hive/pom.xml
index 857affe..e20bdeb 100644
--- a/flink-connectors/flink-connector-hive/pom.xml
+++ b/flink-connectors/flink-connector-hive/pom.xml
@@ -416,6 +416,10 @@ under the License.
<artifactId>hive-hcatalog-core</artifactId>
</exclusion>
<exclusion>
+ <groupId>org.apache.hive.hcatalog</groupId>
+ <artifactId>hive-webhcat-java-client</artifactId>
+ </exclusion>
+ <exclusion>
<groupId>org.apache.tez</groupId>
<artifactId>tez-common</artifactId>
</exclusion>
@@ -601,9 +605,20 @@ under the License.
<artifactId>hadoop-mapreduce-client-core</artifactId>
<groupId>org.apache.hadoop</groupId>
</exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-yarn-server-resourcemanager</artifactId>
+ </exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.hive.hcatalog</groupId>
+ <artifactId>hive-webhcat-java-client</artifactId>
+ <version>${hive.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<!-- TODO: move to flink-connector-hive-test end-to-end test module once it's setup -->
<dependency>
<groupId>org.apache.flink</groupId>
@@ -653,6 +668,14 @@ under the License.
<profiles>
<!-- Activate this profile with -Phive-1.2.1 to build and test against hive-1.2.1 -->
<profile>
+ <id>hive-1.1.1</id>
+ <properties>
+ <hive.version>1.1.1</hive.version>
+ <hivemetastore.hadoop.version>2.6.5</hivemetastore.hadoop.version>
+ <hiverunner.version>3.1.1</hiverunner.version>
+ </properties>
+ </profile>
+ <profile>
<id>hive-1.2.1</id>
<properties>
<hive.version>1.2.1</hive.version>
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java
index 1e66551..49ff7441 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java
@@ -1120,6 +1120,10 @@ public class HiveCatalog extends AbstractCatalog {
public void alterTableStatistics(ObjectPath tablePath, CatalogTableStatistics tableStatistics, boolean ignoreIfNotExists) throws TableNotExistException, CatalogException {
try {
Table hiveTable = getHiveTable(tablePath);
+ // the stats we put in table parameters will be overridden by HMS in older Hive versions, so error out
+ if (!isTablePartitioned(hiveTable) && hiveVersion.compareTo("1.2.1") < 0) {
+ throw new CatalogException("Alter table stats is not supported in Hive version " + hiveVersion);
+ }
// Set table stats
if (compareAndUpdateStatisticsProperties(tableStatistics, hiveTable.getParameters())) {
client.alter_table(tablePath.getDatabaseName(), tablePath.getObjectName(), hiveTable);
@@ -1139,7 +1143,7 @@ public class HiveCatalog extends AbstractCatalog {
Table hiveTable = getHiveTable(tablePath);
// Set table column stats. This only works for non-partitioned tables.
if (!isTablePartitioned(hiveTable)) {
- client.updateTableColumnStatistics(HiveStatsUtil.createTableColumnStats(hiveTable, columnStatistics.getColumnStatisticsData()));
+ client.updateTableColumnStatistics(HiveStatsUtil.createTableColumnStats(hiveTable, columnStatistics.getColumnStatisticsData(), hiveVersion));
} else {
throw new TablePartitionedException(getName(), tablePath);
}
@@ -1204,7 +1208,8 @@ public class HiveCatalog extends AbstractCatalog {
Partition hivePartition = getHivePartition(tablePath, partitionSpec);
Table hiveTable = getHiveTable(tablePath);
String partName = getPartitionName(tablePath, partitionSpec, hiveTable);
- client.updatePartitionColumnStatistics(HiveStatsUtil.createPartitionColumnStats(hivePartition, partName, columnStatistics.getColumnStatisticsData()));
+ client.updatePartitionColumnStatistics(HiveStatsUtil.createPartitionColumnStats(
+ hivePartition, partName, columnStatistics.getColumnStatisticsData(), hiveVersion));
} catch (TableNotExistException | PartitionSpecInvalidException e) {
if (!ignoreIfNotExists) {
throw new PartitionNotExistException(getName(), tablePath, partitionSpec, e);
@@ -1243,7 +1248,7 @@ public class HiveCatalog extends AbstractCatalog {
if (!isTablePartitioned(hiveTable)) {
List<ColumnStatisticsObj> columnStatisticsObjs = client.getTableColumnStatistics(
hiveTable.getDbName(), hiveTable.getTableName(), getFieldNames(hiveTable.getSd().getCols()));
- return new CatalogColumnStatistics(HiveStatsUtil.createCatalogColumnStats(columnStatisticsObjs));
+ return new CatalogColumnStatistics(HiveStatsUtil.createCatalogColumnStats(columnStatisticsObjs, hiveVersion));
} else {
// TableColumnStats of partitioned table is unknown, the behavior is same as HIVE
return CatalogColumnStatistics.UNKNOWN;
@@ -1280,7 +1285,7 @@ public class HiveCatalog extends AbstractCatalog {
getFieldNames(partition.getSd().getCols()));
List<ColumnStatisticsObj> columnStatisticsObjs = partitionColumnStatistics.get(partName);
if (columnStatisticsObjs != null && !columnStatisticsObjs.isEmpty()) {
- return new CatalogColumnStatistics(HiveStatsUtil.createCatalogColumnStats(columnStatisticsObjs));
+ return new CatalogColumnStatistics(HiveStatsUtil.createCatalogColumnStats(columnStatisticsObjs, hiveVersion));
} else {
return CatalogColumnStatistics.UNKNOWN;
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShim.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShim.java
index 1f39d84..2111099 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShim.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShim.java
@@ -18,12 +18,15 @@
package org.apache.flink.table.catalog.hive.client;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataDate;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
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.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.ColumnStatisticsData;
import org.apache.hadoop.hive.metastore.api.Function;
import org.apache.hadoop.hive.metastore.api.InvalidOperationException;
import org.apache.hadoop.hive.metastore.api.MetaException;
@@ -33,16 +36,18 @@ import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.api.UnknownDBException;
import org.apache.hadoop.hive.ql.udf.generic.SimpleGenericUDAFParameterInfo;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
import org.apache.thrift.TException;
import java.io.IOException;
+import java.io.Serializable;
import java.util.List;
import java.util.Map;
/**
* A shim layer to support different versions of Hive.
*/
-public interface HiveShim {
+public interface HiveShim extends Serializable {
/**
* Create a Hive Metastore client based on the given HiveConf object.
@@ -159,4 +164,24 @@ public interface HiveShim {
*/
void makeSpecFromName(Map<String, String> partSpec, Path currPath);
+ /**
+ * Get ObjectInspector for a constant value.
+ */
+ ObjectInspector getObjectInspectorForConstant(PrimitiveTypeInfo primitiveTypeInfo, Object value);
+
+ /**
+ * Generate Hive ColumnStatisticsData from Flink CatalogColumnStatisticsDataDate for DATE columns.
+ */
+ ColumnStatisticsData toHiveDateColStats(CatalogColumnStatisticsDataDate flinkDateColStats);
+
+ /**
+ * Whether a Hive ColumnStatisticsData is for DATE columns.
+ */
+ boolean isDateStats(ColumnStatisticsData colStatsData);
+
+ /**
+ * Generate Flink CatalogColumnStatisticsDataDate from Hive ColumnStatisticsData for DATE columns.
+ */
+ CatalogColumnStatisticsDataDate toFlinkDateColStats(ColumnStatisticsData hiveDateColStats);
+
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimLoader.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimLoader.java
index b3e3903..1645d08 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimLoader.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimLoader.java
@@ -32,6 +32,8 @@ import java.util.concurrent.ConcurrentHashMap;
*/
public class HiveShimLoader {
+ public static final String HIVE_VERSION_V1_1_0 = "1.1.0";
+ public static final String HIVE_VERSION_V1_1_1 = "1.1.1";
public static final String HIVE_VERSION_V1_2_0 = "1.2.0";
public static final String HIVE_VERSION_V1_2_1 = "1.2.1";
public static final String HIVE_VERSION_V1_2_2 = "1.2.2";
@@ -60,6 +62,12 @@ public class HiveShimLoader {
public static HiveShim loadHiveShim(String version) {
return hiveShims.computeIfAbsent(version, (v) -> {
+ if (v.startsWith(HIVE_VERSION_V1_1_0)) {
+ return new HiveShimV110();
+ }
+ if (v.startsWith(HIVE_VERSION_V1_1_1)) {
+ return new HiveShimV111();
+ }
if (v.startsWith(HIVE_VERSION_V1_2_0)) {
return new HiveShimV120();
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV120.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV110.java
similarity index 53%
copy from flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV120.java
copy to flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV110.java
index d1ed444..91fd80a 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV120.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV110.java
@@ -19,6 +19,10 @@
package org.apache.flink.table.catalog.hive.client;
import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataDate;
+import org.apache.flink.table.functions.hive.FlinkHiveUDFException;
+import org.apache.flink.table.functions.hive.conversion.HiveInspectors;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
@@ -26,11 +30,11 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.FileUtils;
import org.apache.hadoop.hive.common.HiveStatsUtils;
-import org.apache.hadoop.hive.common.StatsSetupConst;
import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
-import org.apache.hadoop.hive.metastore.RetryingMetaStoreClient;
import org.apache.hadoop.hive.metastore.Warehouse;
+import org.apache.hadoop.hive.metastore.api.ColumnStatisticsData;
import org.apache.hadoop.hive.metastore.api.Function;
import org.apache.hadoop.hive.metastore.api.InvalidOperationException;
import org.apache.hadoop.hive.metastore.api.MetaException;
@@ -40,6 +44,23 @@ import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.api.UnknownDBException;
import org.apache.hadoop.hive.ql.udf.generic.SimpleGenericUDAFParameterInfo;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantBinaryObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantBooleanObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantByteObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantDateObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantDoubleObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantFloatObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantHiveCharObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantHiveDecimalObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantHiveVarcharObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantIntObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantLongObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantShortObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantStringObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableConstantTimestampObjectInspector;
+import org.apache.hadoop.hive.serde2.typeinfo.CharTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.VarcharTypeInfo;
import org.apache.thrift.TException;
import java.io.IOException;
@@ -51,17 +72,15 @@ import java.util.List;
import java.util.Map;
/**
- * Shim for Hive version 1.2.0.
+ * Shim for Hive version 1.1.0.
*/
-public class HiveShimV120 implements HiveShim {
+public class HiveShimV110 implements HiveShim {
@Override
public IMetaStoreClient getHiveMetastoreClient(HiveConf hiveConf) {
try {
- Method method = RetryingMetaStoreClient.class.getMethod("getProxy", HiveConf.class);
- // getProxy is a static method
- return (IMetaStoreClient) method.invoke(null, (hiveConf));
- } catch (Exception ex) {
+ return new HiveMetaStoreClient(hiveConf);
+ } catch (MetaException ex) {
throw new CatalogException("Failed to create Hive Metastore client", ex);
}
}
@@ -112,9 +131,6 @@ public class HiveShimV120 implements HiveShim {
@Override
public void alterTable(IMetaStoreClient client, String databaseName, String tableName, Table table) throws InvalidOperationException, MetaException, TException {
- // For Hive-1.2.1, we need to tell HMS not to update stats. Otherwise, the stats we put in the table
- // parameters can be overridden. The extra config we add here will be removed by HMS after it's used.
- table.getParameters().put(StatsSetupConst.DO_NOT_UPDATE_STATS, "true");
client.alter_table(databaseName, tableName, table);
}
@@ -194,4 +210,88 @@ public class HiveShimV120 implements HiveShim {
}
}
+ @Override
+ public ObjectInspector getObjectInspectorForConstant(PrimitiveTypeInfo primitiveTypeInfo, Object value) {
+ String className;
+ value = HiveInspectors.hivePrimitiveToWritable(value);
+ // Java constant object inspectors are not available until 1.2.0 -- https://issues.apache.org/jira/browse/HIVE-9766
+ // So we have to use writable constant object inspectors for 1.1.x
+ switch (primitiveTypeInfo.getPrimitiveCategory()) {
+ case BOOLEAN:
+ className = WritableConstantBooleanObjectInspector.class.getName();
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case BYTE:
+ className = WritableConstantByteObjectInspector.class.getName();
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case SHORT:
+ className = WritableConstantShortObjectInspector.class.getName();
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case INT:
+ className = WritableConstantIntObjectInspector.class.getName();
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case LONG:
+ className = WritableConstantLongObjectInspector.class.getName();
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case FLOAT:
+ className = WritableConstantFloatObjectInspector.class.getName();
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case DOUBLE:
+ className = WritableConstantDoubleObjectInspector.class.getName();
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case STRING:
+ className = WritableConstantStringObjectInspector.class.getName();
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case CHAR:
+ className = WritableConstantHiveCharObjectInspector.class.getName();
+ try {
+ return (ObjectInspector) Class.forName(className).getDeclaredConstructor(
+ CharTypeInfo.class, value.getClass()).newInstance(primitiveTypeInfo, value);
+ } catch (Exception e) {
+ throw new FlinkHiveUDFException("Failed to create writable constant object inspector", e);
+ }
+ case VARCHAR:
+ className = WritableConstantHiveVarcharObjectInspector.class.getName();
+ try {
+ return (ObjectInspector) Class.forName(className).getDeclaredConstructor(
+ VarcharTypeInfo.class, value.getClass()).newInstance(primitiveTypeInfo, value);
+ } catch (Exception e) {
+ throw new FlinkHiveUDFException("Failed to create writable constant object inspector", e);
+ }
+ case DATE:
+ className = WritableConstantDateObjectInspector.class.getName();
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case TIMESTAMP:
+ className = WritableConstantTimestampObjectInspector.class.getName();
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case DECIMAL:
+ className = WritableConstantHiveDecimalObjectInspector.class.getName();
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case BINARY:
+ className = WritableConstantBinaryObjectInspector.class.getName();
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case UNKNOWN:
+ case VOID:
+ // If type is null, we use the Constant String to replace
+ className = WritableConstantStringObjectInspector.class.getName();
+ return HiveReflectionUtils.createConstantObjectInspector(className, value.toString());
+ default:
+ throw new FlinkHiveUDFException(
+ String.format("Cannot find ConstantObjectInspector for %s", primitiveTypeInfo));
+ }
+ }
+
+ @Override
+ public ColumnStatisticsData toHiveDateColStats(CatalogColumnStatisticsDataDate flinkDateColStats) {
+ throw new UnsupportedOperationException("DATE column stats are not supported until Hive 1.2.0");
+ }
+
+ @Override
+ public boolean isDateStats(ColumnStatisticsData colStatsData) {
+ return false;
+ }
+
+ @Override
+ public CatalogColumnStatisticsDataDate toFlinkDateColStats(ColumnStatisticsData hiveDateColStats) {
+ throw new UnsupportedOperationException("DATE column stats are not supported until Hive 1.2.0");
+ }
}
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogTestBase.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV111.java
similarity index 53%
copy from flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogTestBase.java
copy to flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV111.java
index 232c250..bafa50a 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogTestBase.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV111.java
@@ -16,30 +16,11 @@
* limitations under the License.
*/
-package org.apache.flink.table.catalog.hive;
-
-import org.apache.flink.table.catalog.CatalogTestBase;
-import org.apache.flink.table.catalog.ObjectPath;
-
-import org.junit.Test;
+package org.apache.flink.table.catalog.hive.client;
/**
- * Base class for testing HiveCatalog.
+ * Shim for Hive version 1.1.1.
*/
-public abstract class HiveCatalogTestBase extends CatalogTestBase {
-
- // ------ functions ------
-
- @Test
- public void testCreateFunctionCaseInsensitive() throws Exception {
- catalog.createDatabase(db1, createDb(), false);
-
- String functionName = "myUdf";
- ObjectPath functionPath = new ObjectPath(db1, functionName);
- catalog.createFunction(functionPath, createFunction(), false);
- // make sure we can get the function
- catalog.getFunction(functionPath);
+public class HiveShimV111 extends HiveShimV110 {
- catalog.dropFunction(functionPath, false);
- }
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV120.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV120.java
index d1ed444..fc92947 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV120.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV120.java
@@ -19,41 +19,30 @@
package org.apache.flink.table.catalog.hive.client;
import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataDate;
+import org.apache.flink.table.catalog.stats.Date;
+import org.apache.flink.table.functions.hive.FlinkHiveUDFException;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hive.common.FileUtils;
-import org.apache.hadoop.hive.common.HiveStatsUtils;
-import org.apache.hadoop.hive.common.StatsSetupConst;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.hadoop.hive.metastore.RetryingMetaStoreClient;
-import org.apache.hadoop.hive.metastore.Warehouse;
-import org.apache.hadoop.hive.metastore.api.Function;
+import org.apache.hadoop.hive.metastore.api.ColumnStatisticsData;
import org.apache.hadoop.hive.metastore.api.InvalidOperationException;
import org.apache.hadoop.hive.metastore.api.MetaException;
-import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
-import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.Table;
-import org.apache.hadoop.hive.metastore.api.UnknownDBException;
-import org.apache.hadoop.hive.ql.udf.generic.SimpleGenericUDAFParameterInfo;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
import org.apache.thrift.TException;
-import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
/**
* Shim for Hive version 1.2.0.
*/
-public class HiveShimV120 implements HiveShim {
+public class HiveShimV120 extends HiveShimV111 {
@Override
public IMetaStoreClient getHiveMetastoreClient(HiveConf hiveConf) {
@@ -67,131 +56,116 @@ public class HiveShimV120 implements HiveShim {
}
@Override
- // 1.x client doesn't support filtering tables by type, so here we need to get all tables and filter by ourselves
- public List<String> getViews(IMetaStoreClient client, String databaseName) throws UnknownDBException, TException {
- // We don't have to use reflection here because client.getAllTables(String) is supposed to be there for
- // all versions.
- List<String> tableNames = client.getAllTables(databaseName);
- List<String> views = new ArrayList<>();
- for (String name : tableNames) {
- Table table = client.getTable(databaseName, name);
- String viewDef = table.getViewOriginalText();
- if (viewDef != null && !viewDef.isEmpty()) {
- views.add(table.getTableName());
- }
- }
- return views;
- }
-
- @Override
- public Function getFunction(IMetaStoreClient client, String dbName, String functionName) throws NoSuchObjectException, TException {
- try {
- // hive-1.x doesn't throw NoSuchObjectException if function doesn't exist, instead it throws a MetaException
- return client.getFunction(dbName, functionName);
- } catch (MetaException e) {
- // need to check the cause and message of this MetaException to decide whether it should actually be a NoSuchObjectException
- if (e.getCause() instanceof NoSuchObjectException) {
- throw (NoSuchObjectException) e.getCause();
- }
- if (e.getMessage().startsWith(NoSuchObjectException.class.getSimpleName())) {
- throw new NoSuchObjectException(e.getMessage());
- }
- throw e;
- }
- }
-
- @Override
- public boolean moveToTrash(FileSystem fs, Path path, Configuration conf, boolean purge) throws IOException {
- try {
- Method method = FileUtils.class.getDeclaredMethod("moveToTrash", FileSystem.class, Path.class, Configuration.class);
- return (boolean) method.invoke(null, fs, path, conf);
- } catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) {
- throw new IOException("Failed to move " + path + " to trash", e);
- }
- }
-
- @Override
public void alterTable(IMetaStoreClient client, String databaseName, String tableName, Table table) throws InvalidOperationException, MetaException, TException {
- // For Hive-1.2.1, we need to tell HMS not to update stats. Otherwise, the stats we put in the table
+ // For Hive-1.2.x, we need to tell HMS not to update stats. Otherwise, the stats we put in the table
// parameters can be overridden. The extra config we add here will be removed by HMS after it's used.
- table.getParameters().put(StatsSetupConst.DO_NOT_UPDATE_STATS, "true");
+ // Don't use StatsSetupConst.DO_NOT_UPDATE_STATS because it wasn't defined in Hive 1.1.x.
+ table.getParameters().put("DO_NOT_UPDATE_STATS", "true");
client.alter_table(databaseName, tableName, table);
}
@Override
- public void alterPartition(IMetaStoreClient client, String databaseName, String tableName, Partition partition)
- throws InvalidOperationException, MetaException, TException {
- String errorMsg = "Failed to alter partition for table %s in database %s";
- try {
- Method method = client.getClass().getMethod("alter_partition", String.class, String.class, Partition.class);
- method.invoke(client, databaseName, tableName, partition);
- } catch (InvocationTargetException ite) {
- Throwable targetEx = ite.getTargetException();
- if (targetEx instanceof TException) {
- throw (TException) targetEx;
- } else {
- throw new CatalogException(String.format(errorMsg, tableName, databaseName), targetEx);
- }
- } catch (NoSuchMethodException | IllegalAccessException e) {
- throw new CatalogException(String.format(errorMsg, tableName, databaseName), e);
- }
- }
-
- @Override
- public SimpleGenericUDAFParameterInfo createUDAFParameterInfo(ObjectInspector[] params, boolean isWindowing, boolean distinct, boolean allColumns) {
- try {
- Constructor constructor = SimpleGenericUDAFParameterInfo.class.getConstructor(ObjectInspector[].class,
- boolean.class, boolean.class);
- return (SimpleGenericUDAFParameterInfo) constructor.newInstance(params, distinct, allColumns);
- } catch (NoSuchMethodException | IllegalAccessException | InstantiationException | InvocationTargetException e) {
- throw new CatalogException("Failed to create SimpleGenericUDAFParameterInfo", e);
+ public ObjectInspector getObjectInspectorForConstant(PrimitiveTypeInfo primitiveTypeInfo, Object value) {
+ String className;
+ switch (primitiveTypeInfo.getPrimitiveCategory()) {
+ case BOOLEAN:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantBooleanObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case BYTE:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantByteObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case SHORT:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantShortObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case INT:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantIntObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case LONG:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantLongObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case FLOAT:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantFloatObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case DOUBLE:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantDoubleObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case STRING:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantStringObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case CHAR:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantHiveCharObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case VARCHAR:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantHiveVarcharObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case DATE:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantDateObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case TIMESTAMP:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantTimestampObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case DECIMAL:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantHiveDecimalObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case BINARY:
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantBinaryObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value);
+ case UNKNOWN:
+ case VOID:
+ // If type is null, we use the Java Constant String to replace
+ className = "org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantStringObjectInspector";
+ return HiveReflectionUtils.createConstantObjectInspector(className, value.toString());
+ default:
+ throw new FlinkHiveUDFException(
+ String.format("Cannot find ConstantObjectInspector for %s", primitiveTypeInfo));
}
}
@Override
- public Class<?> getMetaStoreUtilsClass() {
+ public ColumnStatisticsData toHiveDateColStats(CatalogColumnStatisticsDataDate flinkDateColStats) {
try {
- return Class.forName("org.apache.hadoop.hive.metastore.MetaStoreUtils");
- } catch (ClassNotFoundException e) {
- throw new CatalogException("Failed to find class MetaStoreUtils", e);
+ Class dateStatsClz = Class.forName("org.apache.hadoop.hive.metastore.api.DateColumnStatsData");
+ Object dateStats = dateStatsClz.getDeclaredConstructor(long.class, long.class)
+ .newInstance(flinkDateColStats.getNullCount(), flinkDateColStats.getNdv());
+ Class hmsDateClz = Class.forName("org.apache.hadoop.hive.metastore.api.Date");
+ Method setHigh = dateStatsClz.getDeclaredMethod("setHighValue", hmsDateClz);
+ Method setLow = dateStatsClz.getDeclaredMethod("setLowValue", hmsDateClz);
+ Constructor hmsDateConstructor = hmsDateClz.getConstructor(long.class);
+ setHigh.invoke(dateStats, hmsDateConstructor.newInstance(flinkDateColStats.getMax().getDaysSinceEpoch()));
+ setLow.invoke(dateStats, hmsDateConstructor.newInstance(flinkDateColStats.getMin().getDaysSinceEpoch()));
+ Class colStatsClz = ColumnStatisticsData.class;
+ return (ColumnStatisticsData) colStatsClz.getDeclaredMethod("dateStats", dateStatsClz).invoke(null, dateStats);
+ } catch (ClassNotFoundException | NoSuchMethodException | InstantiationException | IllegalAccessException | InvocationTargetException e) {
+ throw new CatalogException("Failed to create Hive statistics for date column", e);
}
}
@Override
- public Class<?> getHiveMetaStoreUtilsClass() {
- return getMetaStoreUtilsClass();
- }
-
- @Override
- public Class<?> getDateDataTypeClass() {
- return java.sql.Date.class;
- }
-
- @Override
- public Class<?> getTimestampDataTypeClass() {
- return java.sql.Timestamp.class;
- }
-
- @Override
- public FileStatus[] getFileStatusRecurse(Path path, int level, FileSystem fs) throws IOException {
+ public boolean isDateStats(ColumnStatisticsData colStatsData) {
try {
- Method method = HiveStatsUtils.class.getMethod("getFileStatusRecurse", Path.class, Integer.TYPE, FileSystem.class);
- // getFileStatusRecurse is a static method
- return (FileStatus[]) method.invoke(null, path, level, fs);
- } catch (Exception ex) {
- throw new CatalogException("Failed to invoke HiveStatsUtils.getFileStatusRecurse()", ex);
+ Method method = ColumnStatisticsData.class.getDeclaredMethod("isSetDateStats");
+ return (boolean) method.invoke(colStatsData);
+ } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
+ throw new CatalogException("Failed to decide whether ColumnStatisticsData is for DATE column", e);
}
}
@Override
- public void makeSpecFromName(Map<String, String> partSpec, Path currPath) {
+ public CatalogColumnStatisticsDataDate toFlinkDateColStats(ColumnStatisticsData hiveDateColStats) {
try {
- Method method = Warehouse.class.getMethod("makeSpecFromName", Map.class, Path.class);
- // makeSpecFromName is a static method
- method.invoke(null, partSpec, currPath);
- } catch (Exception ex) {
- throw new CatalogException("Failed to invoke Warehouse.makeSpecFromName()", ex);
+ Object dateStats = ColumnStatisticsData.class.getDeclaredMethod("getDateStats").invoke(hiveDateColStats);
+ Class dateStatsClz = dateStats.getClass();
+ long numDV = (long) dateStatsClz.getMethod("getNumDVs").invoke(dateStats);
+ long numNull = (long) dateStatsClz.getMethod("getNumNulls").invoke(dateStats);
+ Object hmsHighDate = dateStatsClz.getMethod("getHighValue").invoke(dateStats);
+ Object hmsLowDate = dateStatsClz.getMethod("getLowValue").invoke(dateStats);
+ Class hmsDateClz = hmsHighDate.getClass();
+ Method hmsDateDays = hmsDateClz.getMethod("getDaysSinceEpoch");
+ long highDateDays = (long) hmsDateDays.invoke(hmsHighDate);
+ long lowDateDays = (long) hmsDateDays.invoke(hmsLowDate);
+ return new CatalogColumnStatisticsDataDate(new Date(lowDateDays), new Date(highDateDays), numDV, numNull);
+ } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
+ throw new CatalogException("Failed to create Flink statistics for date column", e);
}
}
-
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveReflectionUtils.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveReflectionUtils.java
index 1941096..b03eb66 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveReflectionUtils.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveReflectionUtils.java
@@ -28,8 +28,7 @@ import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.serde2.Deserializer;
import org.apache.hadoop.hive.serde2.SerDeException;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantDateObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantTimestampObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
@@ -88,25 +87,14 @@ public class HiveReflectionUtils {
}
}
- public static JavaConstantDateObjectInspector createJavaConstantDateObjectInspector(HiveShim hiveShim, Object value) {
- Constructor<?> meth = null;
+ public static ObjectInspector createConstantObjectInspector(String className, Object value) {
try {
- meth = JavaConstantDateObjectInspector.class.getDeclaredConstructor(hiveShim.getDateDataTypeClass());
- meth.setAccessible(true);
- return (JavaConstantDateObjectInspector) meth.newInstance(value);
- } catch (NoSuchMethodException | InstantiationException | IllegalAccessException | InvocationTargetException e) {
- throw new FlinkHiveUDFException("Failed to instantiate JavaConstantDateObjectInspector");
- }
- }
-
- public static JavaConstantTimestampObjectInspector createJavaConstantTimestampObjectInspector(HiveShim hiveShim, Object value) {
- Constructor<?> meth = null;
- try {
- meth = JavaConstantTimestampObjectInspector.class.getDeclaredConstructor(hiveShim.getDateDataTypeClass());
- meth.setAccessible(true);
- return (JavaConstantTimestampObjectInspector) meth.newInstance(value);
- } catch (NoSuchMethodException | InstantiationException | IllegalAccessException | InvocationTargetException e) {
- throw new FlinkHiveUDFException("Failed to instantiate JavaConstantTimestampObjectInspector");
+ Constructor<?> method = Class.forName(className).getDeclaredConstructor(value.getClass());
+ method.setAccessible(true);
+ return (ObjectInspector) method.newInstance(value);
+ } catch (ClassNotFoundException | NoSuchMethodException | InstantiationException | IllegalAccessException
+ | InvocationTargetException e) {
+ throw new FlinkHiveUDFException("Failed to instantiate JavaConstantDateObjectInspector", e);
}
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveStatsUtil.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveStatsUtil.java
index 232b5f6..7f0551f 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveStatsUtil.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveStatsUtil.java
@@ -19,6 +19,8 @@
package org.apache.flink.table.catalog.hive.util;
import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataBase;
import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataBinary;
import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataBoolean;
@@ -35,7 +37,6 @@ import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsData;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsDesc;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
-import org.apache.hadoop.hive.metastore.api.DateColumnStatsData;
import org.apache.hadoop.hive.metastore.api.DoubleColumnStatsData;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.LongColumnStatsData;
@@ -69,13 +70,13 @@ public class HiveStatsUtil {
/**
* Create a map of Flink column stats from the given Hive column stats.
*/
- public static Map<String, CatalogColumnStatisticsDataBase> createCatalogColumnStats(@Nonnull List<ColumnStatisticsObj> hiveColStats) {
+ public static Map<String, CatalogColumnStatisticsDataBase> createCatalogColumnStats(@Nonnull List<ColumnStatisticsObj> hiveColStats, String hiveVersion) {
checkNotNull(hiveColStats, "hiveColStats can not be null");
Map<String, CatalogColumnStatisticsDataBase> colStats = new HashMap<>();
for (ColumnStatisticsObj colStatsObj : hiveColStats) {
CatalogColumnStatisticsDataBase columnStats = createTableColumnStats(
HiveTypeUtil.toFlinkType(TypeInfoUtils.getTypeInfoFromTypeString(colStatsObj.getColType())),
- colStatsObj.getStatsData());
+ colStatsObj.getStatsData(), hiveVersion);
colStats.put(colStatsObj.getColName(), columnStats);
}
@@ -87,9 +88,10 @@ public class HiveStatsUtil {
*/
public static ColumnStatistics createTableColumnStats(
Table hiveTable,
- Map<String, CatalogColumnStatisticsDataBase> colStats) {
+ Map<String, CatalogColumnStatisticsDataBase> colStats,
+ String hiveVersion) {
ColumnStatisticsDesc desc = new ColumnStatisticsDesc(true, hiveTable.getDbName(), hiveTable.getTableName());
- return createHiveColumnStatistics(colStats, hiveTable.getSd(), desc);
+ return createHiveColumnStatistics(colStats, hiveTable.getSd(), desc, hiveVersion);
}
/**
@@ -98,16 +100,18 @@ public class HiveStatsUtil {
public static ColumnStatistics createPartitionColumnStats(
Partition hivePartition,
String partName,
- Map<String, CatalogColumnStatisticsDataBase> colStats) {
+ Map<String, CatalogColumnStatisticsDataBase> colStats,
+ String hiveVersion) {
ColumnStatisticsDesc desc = new ColumnStatisticsDesc(false, hivePartition.getDbName(), hivePartition.getTableName());
desc.setPartName(partName);
- return createHiveColumnStatistics(colStats, hivePartition.getSd(), desc);
+ return createHiveColumnStatistics(colStats, hivePartition.getSd(), desc, hiveVersion);
}
private static ColumnStatistics createHiveColumnStatistics(
Map<String, CatalogColumnStatisticsDataBase> colStats,
StorageDescriptor sd,
- ColumnStatisticsDesc desc) {
+ ColumnStatisticsDesc desc,
+ String hiveVersion) {
List<ColumnStatisticsObj> colStatsList = new ArrayList<>();
for (FieldSchema field : sd.getCols()) {
@@ -115,8 +119,10 @@ public class HiveStatsUtil {
String hiveColType = field.getType();
CatalogColumnStatisticsDataBase flinkColStat = colStats.get(field.getName());
if (null != flinkColStat) {
- ColumnStatisticsData statsData =
- getColumnStatisticsData(HiveTypeUtil.toFlinkType(TypeInfoUtils.getTypeInfoFromTypeString(hiveColType)), flinkColStat);
+ ColumnStatisticsData statsData = getColumnStatisticsData(
+ HiveTypeUtil.toFlinkType(TypeInfoUtils.getTypeInfoFromTypeString(hiveColType)),
+ flinkColStat,
+ hiveVersion);
ColumnStatisticsObj columnStatisticsObj = new ColumnStatisticsObj(hiveColName, hiveColType, statsData);
colStatsList.add(columnStatisticsObj);
}
@@ -128,7 +134,8 @@ public class HiveStatsUtil {
/**
* Create Flink ColumnStats from Hive ColumnStatisticsData.
*/
- private static CatalogColumnStatisticsDataBase createTableColumnStats(DataType colType, ColumnStatisticsData stats) {
+ private static CatalogColumnStatisticsDataBase createTableColumnStats(DataType colType, ColumnStatisticsData stats, String hiveVersion) {
+ HiveShim hiveShim = HiveShimLoader.loadHiveShim(hiveVersion);
if (stats.isSetBinaryStats()) {
BinaryColumnStatsData binaryStats = stats.getBinaryStats();
return new CatalogColumnStatisticsDataBinary(
@@ -141,13 +148,8 @@ public class HiveStatsUtil {
booleanStats.getNumTrues(),
booleanStats.getNumFalses(),
booleanStats.getNumNulls());
- } else if (stats.isSetDateStats()) {
- DateColumnStatsData dateStats = stats.getDateStats();
- return new CatalogColumnStatisticsDataDate(
- new org.apache.flink.table.catalog.stats.Date(dateStats.getLowValue().getDaysSinceEpoch()),
- new org.apache.flink.table.catalog.stats.Date(dateStats.getHighValue().getDaysSinceEpoch()),
- dateStats.getNumDVs(),
- dateStats.getNumNulls());
+ } else if (hiveShim.isDateStats(stats)) {
+ return hiveShim.toFlinkDateColStats(stats);
} else if (stats.isSetDoubleStats()) {
DoubleColumnStatsData doubleStats = stats.getDoubleStats();
return new CatalogColumnStatisticsDataDouble(
@@ -180,7 +182,8 @@ public class HiveStatsUtil {
* Note we currently assume that, in Flink, the max and min of ColumnStats will be same type as the Flink column type.
* For example, for SHORT and Long columns, the max and min of their ColumnStats should be of type SHORT and LONG.
*/
- private static ColumnStatisticsData getColumnStatisticsData(DataType colType, CatalogColumnStatisticsDataBase colStat) {
+ private static ColumnStatisticsData getColumnStatisticsData(DataType colType, CatalogColumnStatisticsDataBase colStat,
+ String hiveVersion) {
LogicalTypeRoot type = colType.getLogicalType().getTypeRoot();
if (type.equals(LogicalTypeRoot.CHAR)
|| type.equals(LogicalTypeRoot.VARCHAR)) {
@@ -222,11 +225,8 @@ public class HiveStatsUtil {
}
} else if (type.equals(LogicalTypeRoot.DATE)) {
if (colStat instanceof CatalogColumnStatisticsDataDate) {
- CatalogColumnStatisticsDataDate dateColumnStatsData = (CatalogColumnStatisticsDataDate) colStat;
- DateColumnStatsData dateStats = new DateColumnStatsData(dateColumnStatsData.getNullCount(), dateColumnStatsData.getNdv());
- dateStats.setHighValue(new org.apache.hadoop.hive.metastore.api.Date(dateColumnStatsData.getMax().getDaysSinceEpoch()));
- dateStats.setLowValue(new org.apache.hadoop.hive.metastore.api.Date(dateColumnStatsData.getMin().getDaysSinceEpoch()));
- return ColumnStatisticsData.dateStats(dateStats);
+ HiveShim hiveShim = HiveShimLoader.loadHiveShim(hiveVersion);
+ return hiveShim.toHiveDateColStats((CatalogColumnStatisticsDataDate) colStat);
}
} else if (type.equals(LogicalTypeRoot.VARBINARY)
|| type.equals(LogicalTypeRoot.BINARY)) {
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTypeUtil.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTypeUtil.java
index 68c3ede..b5f7dac 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTypeUtil.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTypeUtil.java
@@ -323,4 +323,13 @@ public class HiveTypeUtil {
String.format("Flink doesn't support converting type %s to Hive type yet.", dataType.toString()));
}
}
+
+ /**
+ * INTERVAL are not available in older versions. So better to have our own enum for primitive categories.
+ */
+ public enum HivePrimitiveCategory {
+ VOID, BOOLEAN, BYTE, SHORT, INT, LONG, FLOAT, DOUBLE, STRING,
+ DATE, TIMESTAMP, BINARY, DECIMAL, VARCHAR, CHAR, INTERVAL_YEAR_MONTH, INTERVAL_DAY_TIME,
+ UNKNOWN
+ }
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveGenericUDF.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveGenericUDF.java
index e970ace..5b1ebda 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveGenericUDF.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveGenericUDF.java
@@ -41,7 +41,7 @@ public class HiveGenericUDF extends HiveScalarFunction<GenericUDF> {
private static final Logger LOG = LoggerFactory.getLogger(HiveGenericUDF.class);
private transient GenericUDF.DeferredObject[] deferredObjects;
- private transient HiveShim hiveShim;
+ private HiveShim hiveShim;
public HiveGenericUDF(HiveFunctionWrapper<GenericUDF> hiveFunctionWrapper, HiveShim hiveShim) {
super(hiveFunctionWrapper);
@@ -56,9 +56,10 @@ public class HiveGenericUDF extends HiveScalarFunction<GenericUDF> {
function = hiveFunctionWrapper.createFunction();
+ ObjectInspector[] argInspectors = HiveInspectors.toInspectors(hiveShim, constantArguments, argTypes);
+
try {
- returnInspector = function.initializeAndFoldConstants(
- HiveInspectors.toInspectors(hiveShim, constantArguments, argTypes));
+ returnInspector = function.initializeAndFoldConstants(argInspectors);
} catch (UDFArgumentException e) {
throw new FlinkHiveUDFException(e);
}
@@ -67,8 +68,7 @@ public class HiveGenericUDF extends HiveScalarFunction<GenericUDF> {
for (int i = 0; i < deferredObjects.length; i++) {
deferredObjects[i] = new DeferredObjectAdapter(
- TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(
- HiveTypeUtil.toHiveTypeInfo(argTypes[i])),
+ argInspectors[i],
argTypes[i].getLogicalType()
);
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveGenericUDTF.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveGenericUDTF.java
index a7dbc8b..3578a83 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveGenericUDTF.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveGenericUDTF.java
@@ -64,7 +64,7 @@ public class HiveGenericUDTF extends TableFunction<Row> implements HiveFunction
private transient boolean allIdentityConverter;
private transient HiveObjectConversion[] conversions;
- private transient HiveShim hiveShim;
+ private HiveShim hiveShim;
public HiveGenericUDTF(HiveFunctionWrapper<GenericUDTF> hiveFunctionWrapper, HiveShim hiveShim) {
this.hiveFunctionWrapper = hiveFunctionWrapper;
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/conversion/HiveInspectors.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/conversion/HiveInspectors.java
index e7e88b2..377e1e9 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/conversion/HiveInspectors.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/conversion/HiveInspectors.java
@@ -22,7 +22,6 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.hive.client.HiveShim;
-import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
import org.apache.flink.table.catalog.hive.util.HiveTypeUtil;
import org.apache.flink.table.functions.hive.FlinkHiveUDFException;
import org.apache.flink.table.types.DataType;
@@ -45,7 +44,6 @@ import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
import org.apache.hadoop.hive.serde2.io.HiveVarcharWritable;
import org.apache.hadoop.hive.serde2.io.ShortWritable;
import org.apache.hadoop.hive.serde2.io.TimestampWritable;
-import org.apache.hadoop.hive.serde2.objectinspector.ConstantObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ListObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.MapObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
@@ -64,18 +62,6 @@ import org.apache.hadoop.hive.serde2.objectinspector.primitive.HiveCharObjectIns
import org.apache.hadoop.hive.serde2.objectinspector.primitive.HiveDecimalObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.HiveVarcharObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.IntObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantBinaryObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantBooleanObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantByteObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantDoubleObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantFloatObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantHiveCharObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantHiveDecimalObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantHiveVarcharObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantIntObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantLongObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantShortObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.JavaConstantStringObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.LongObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.ShortObjectInspector;
@@ -95,10 +81,12 @@ import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
import java.lang.reflect.Array;
import java.math.BigDecimal;
import java.sql.Date;
+import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -130,64 +118,20 @@ public class HiveInspectors {
TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(
HiveTypeUtil.toHiveTypeInfo(argTypes[i]));
} else {
- argumentInspectors[i] =
- HiveInspectors.getPrimitiveJavaConstantObjectInspector(
- hiveShim,
- (PrimitiveTypeInfo) HiveTypeUtil.toHiveTypeInfo(argTypes[i]),
- constant
- );
+ PrimitiveTypeInfo primitiveTypeInfo = (PrimitiveTypeInfo) HiveTypeUtil.toHiveTypeInfo(argTypes[i]);
+ argumentInspectors[i] = hiveShim.getObjectInspectorForConstant(primitiveTypeInfo, constant);
}
}
return argumentInspectors;
}
- private static ConstantObjectInspector getPrimitiveJavaConstantObjectInspector(HiveShim hiveShim,
- PrimitiveTypeInfo typeInfo, Object value) {
- switch (typeInfo.getPrimitiveCategory()) {
- case BOOLEAN:
- return new JavaConstantBooleanObjectInspector((Boolean) value);
- case BYTE:
- return new JavaConstantByteObjectInspector((Byte) value);
- case SHORT:
- return new JavaConstantShortObjectInspector((Short) value);
- case INT:
- return new JavaConstantIntObjectInspector((Integer) value);
- case LONG:
- return new JavaConstantLongObjectInspector((Long) value);
- case FLOAT:
- return new JavaConstantFloatObjectInspector((Float) value);
- case DOUBLE:
- return new JavaConstantDoubleObjectInspector((Double) value);
- case STRING:
- return new JavaConstantStringObjectInspector((String) value);
- case CHAR:
- return new JavaConstantHiveCharObjectInspector((HiveChar) value);
- case VARCHAR:
- return new JavaConstantHiveVarcharObjectInspector((HiveVarchar) value);
- case DATE:
- return HiveReflectionUtils.createJavaConstantDateObjectInspector(hiveShim, value);
- case TIMESTAMP:
- return HiveReflectionUtils.createJavaConstantTimestampObjectInspector(hiveShim, value);
- case DECIMAL:
- return new JavaConstantHiveDecimalObjectInspector((HiveDecimal) value);
- case BINARY:
- return new JavaConstantBinaryObjectInspector((byte[]) value);
- case UNKNOWN:
- case VOID:
- // If type is null, we use the Java Constant String to replace
- return new JavaConstantStringObjectInspector((String) value);
- default:
- throw new FlinkHiveUDFException(
- String.format("Cannot find ConstantObjectInspector for %s", typeInfo));
- }
- }
-
/**
* Get conversion for converting Flink object to Hive object from an ObjectInspector and the corresponding Flink DataType.
*/
public static HiveObjectConversion getConversion(ObjectInspector inspector, LogicalType dataType) {
if (inspector instanceof PrimitiveObjectInspector) {
+ HiveObjectConversion conversion;
if (inspector instanceof BooleanObjectInspector ||
inspector instanceof StringObjectInspector ||
inspector instanceof ByteObjectInspector ||
@@ -199,14 +143,20 @@ public class HiveInspectors {
inspector instanceof DateObjectInspector ||
inspector instanceof TimestampObjectInspector ||
inspector instanceof BinaryObjectInspector) {
- return IdentityConversion.INSTANCE;
+ conversion = IdentityConversion.INSTANCE;
} else if (inspector instanceof HiveCharObjectInspector) {
- return o -> new HiveChar((String) o, ((CharType) dataType).getLength());
+ conversion = o -> new HiveChar((String) o, ((CharType) dataType).getLength());
} else if (inspector instanceof HiveVarcharObjectInspector) {
- return o -> new HiveVarchar((String) o, ((VarCharType) dataType).getLength());
+ conversion = o -> new HiveVarchar((String) o, ((VarCharType) dataType).getLength());
} else if (inspector instanceof HiveDecimalObjectInspector) {
- return o -> o == null ? null : HiveDecimal.create((BigDecimal) o);
+ conversion = o -> o == null ? null : HiveDecimal.create((BigDecimal) o);
+ } else {
+ throw new FlinkHiveUDFException("Unsupported primitive object inspector " + inspector.getClass().getName());
+ }
+ if (((PrimitiveObjectInspector) inspector).preferWritable()) {
+ conversion = new WritableHiveObjectConversion(conversion);
}
+ return conversion;
}
if (inspector instanceof ListObjectInspector) {
@@ -460,7 +410,7 @@ public class HiveInspectors {
switch (inspector.getCategory()) {
case PRIMITIVE: {
PrimitiveObjectInspector primitiveOI = (PrimitiveObjectInspector) inspector;
- switch (primitiveOI.getPrimitiveCategory()) {
+ switch (HiveTypeUtil.HivePrimitiveCategory.valueOf(primitiveOI.getPrimitiveCategory().name())) {
case STRING:
case CHAR:
case VARCHAR:
@@ -506,4 +456,47 @@ public class HiveInspectors {
throw new IllegalArgumentException("Unsupported type " + inspector.getCategory().name());
}
}
+
+ /**
+ * Converts a Hive primitive java object to corresponding Writable object.
+ */
+ public static Writable hivePrimitiveToWritable(Object value) {
+ Writable writable;
+ // in case value is already a Writable
+ if (value instanceof Writable) {
+ writable = (Writable) value;
+ } else if (value instanceof Boolean) {
+ writable = new BooleanWritable((Boolean) value);
+ } else if (value instanceof Byte) {
+ writable = new ByteWritable((Byte) value);
+ } else if (value instanceof Short) {
+ writable = new ShortWritable((Short) value);
+ } else if (value instanceof Integer) {
+ writable = new IntWritable((Integer) value);
+ } else if (value instanceof Long) {
+ writable = new LongWritable((Long) value);
+ } else if (value instanceof Float) {
+ writable = new FloatWritable((Float) value);
+ } else if (value instanceof Double) {
+ writable = new DoubleWritable((Double) value);
+ } else if (value instanceof String) {
+ writable = new Text((String) value);
+ } else if (value instanceof HiveChar) {
+ writable = new HiveCharWritable((HiveChar) value);
+ } else if (value instanceof HiveVarchar) {
+ writable = new HiveVarcharWritable((HiveVarchar) value);
+ } else if (value instanceof Date) {
+ writable = new DateWritable((Date) value);
+ } else if (value instanceof Timestamp) {
+ writable = new TimestampWritable((Timestamp) value);
+ } else if (value instanceof BigDecimal) {
+ HiveDecimal hiveDecimal = HiveDecimal.create((BigDecimal) value);
+ writable = new HiveDecimalWritable(hiveDecimal);
+ } else if (value instanceof byte[]) {
+ writable = new BytesWritable((byte[]) value);
+ } else {
+ throw new CatalogException("Unsupported primitive java value of class " + value.getClass().getName());
+ }
+ return writable;
+ }
}
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogTestBase.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/conversion/WritableHiveObjectConversion.java
similarity index 53%
copy from flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogTestBase.java
copy to flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/conversion/WritableHiveObjectConversion.java
index 232c250..c659569 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogTestBase.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/conversion/WritableHiveObjectConversion.java
@@ -16,30 +16,21 @@
* limitations under the License.
*/
-package org.apache.flink.table.catalog.hive;
-
-import org.apache.flink.table.catalog.CatalogTestBase;
-import org.apache.flink.table.catalog.ObjectPath;
-
-import org.junit.Test;
+package org.apache.flink.table.functions.hive.conversion;
/**
- * Base class for testing HiveCatalog.
+ * A HiveObjectConversion that converts Flink objects to Hive Writable objects.
*/
-public abstract class HiveCatalogTestBase extends CatalogTestBase {
+public class WritableHiveObjectConversion implements HiveObjectConversion {
- // ------ functions ------
+ private final HiveObjectConversion flinkToJavaConversion;
- @Test
- public void testCreateFunctionCaseInsensitive() throws Exception {
- catalog.createDatabase(db1, createDb(), false);
-
- String functionName = "myUdf";
- ObjectPath functionPath = new ObjectPath(db1, functionName);
- catalog.createFunction(functionPath, createFunction(), false);
- // make sure we can get the function
- catalog.getFunction(functionPath);
+ WritableHiveObjectConversion(HiveObjectConversion flinkToJavaConversion) {
+ this.flinkToJavaConversion = flinkToJavaConversion;
+ }
- catalog.dropFunction(functionPath, false);
+ @Override
+ public Object toHiveObject(Object o) {
+ return HiveInspectors.hivePrimitiveToWritable(flinkToJavaConversion.toHiveObject(o));
}
}
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/FlinkStandaloneHiveServerContext.java
index 236f253..6c90da8 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/FlinkStandaloneHiveServerContext.java
@@ -107,6 +107,9 @@ public class FlinkStandaloneHiveServerContext implements HiveServerContext {
hiveConf.setBoolVar(HIVE_SERVER2_LOGGING_OPERATION_ENABLED, false);
hiveConf.setVar(HADOOPBIN, "NO_BIN!");
+
+ // To avoid https://issues.apache.org/jira/browse/HIVE-13185 when loading data into tables
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVECHECKFILEFORMAT, false);
}
private void overrideHiveConf(HiveConf hiveConf) {
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveRunnerShimLoader.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveRunnerShimLoader.java
index ba2ff57..2a344d1 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveRunnerShimLoader.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveRunnerShimLoader.java
@@ -37,6 +37,8 @@ public class HiveRunnerShimLoader {
String hiveVersion = HiveShimLoader.getHiveVersion();
return hiveRunnerShims.computeIfAbsent(hiveVersion, v -> {
switch (v) {
+ case HiveShimLoader.HIVE_VERSION_V1_1_0:
+ case HiveShimLoader.HIVE_VERSION_V1_1_1:
case HiveShimLoader.HIVE_VERSION_V1_2_0:
case HiveShimLoader.HIVE_VERSION_V1_2_1:
case HiveShimLoader.HIVE_VERSION_V1_2_2:
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceTest.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceTest.java
index 878cd68..8a8aa0f 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceTest.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceTest.java
@@ -84,12 +84,11 @@ public class HiveTableSourceTest {
final String dbName = "source_db";
final String tblName = "test";
hiveShell.execute("CREATE TABLE source_db.test ( a INT, b INT, c STRING, d BIGINT, e DOUBLE)");
- hiveShell.insertInto(dbName, tblName)
- .withAllColumns()
- .addRow(1, 1, "a", 1000L, 1.11)
- .addRow(2, 2, "b", 2000L, 2.22)
- .addRow(3, 3, "c", 3000L, 3.33)
- .addRow(4, 4, "d", 4000L, 4.44)
+ HiveTestUtils.createTextTableInserter(hiveShell, 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();
TableEnvironment tEnv = HiveTestUtils.createTableEnv();
@@ -116,9 +115,8 @@ public class HiveTableSourceTest {
map.put(1, "a");
map.put(2, "b");
Object[] struct = new Object[]{3, 3L};
- hiveShell.insertInto(dbName, tblName)
- .withAllColumns()
- .addRow(array, map, struct)
+ HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ .addRow(new Object[]{array, map, struct})
.commit();
TableEnvironment tEnv = HiveTestUtils.createTableEnv();
tEnv.registerCatalog(catalogName, hiveCatalog);
@@ -141,13 +139,14 @@ public class HiveTableSourceTest {
final String tblName = "test_table_pt";
hiveShell.execute("CREATE TABLE source_db.test_table_pt " +
"(year STRING, value INT) partitioned by (pt int);");
- hiveShell.insertInto(dbName, tblName)
- .withColumns("year", "value", "pt")
- .addRow("2014", 3, 0)
- .addRow("2014", 4, 0)
- .addRow("2015", 2, 1)
- .addRow("2015", 5, 1)
- .commit();
+ HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ .addRow(new Object[]{"2014", 3})
+ .addRow(new Object[]{"2014", 4})
+ .commit("pt=0");
+ HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ .addRow(new Object[]{"2015", 2})
+ .addRow(new Object[]{"2015", 5})
+ .commit("pt=1");
TableEnvironment tEnv = HiveTestUtils.createTableEnv();
tEnv.registerCatalog(catalogName, hiveCatalog);
Table src = tEnv.sqlQuery("select * from hive.source_db.test_table_pt");
@@ -165,13 +164,14 @@ public class HiveTableSourceTest {
final String tblName = "test_table_pt_1";
hiveShell.execute("CREATE TABLE source_db.test_table_pt_1 " +
"(year STRING, value INT) partitioned by (pt int);");
- hiveShell.insertInto(dbName, tblName)
- .withColumns("year", "value", "pt")
- .addRow("2014", 3, 0)
- .addRow("2014", 4, 0)
- .addRow("2015", 2, 1)
- .addRow("2015", 5, 1)
- .commit();
+ HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ .addRow(new Object[]{"2014", 3})
+ .addRow(new Object[]{"2014", 4})
+ .commit("pt=0");
+ HiveTestUtils.createTextTableInserter(hiveShell, dbName, tblName)
+ .addRow(new Object[]{"2015", 2})
+ .addRow(new Object[]{"2015", 5})
+ .commit("pt=1");
TableEnvironment tEnv = HiveTestUtils.createTableEnv();
tEnv.registerCatalog(catalogName, hiveCatalog);
Table src = tEnv.sqlQuery("select * from hive.source_db.test_table_pt_1 where pt = 0");
@@ -196,11 +196,13 @@ public class HiveTableSourceTest {
hiveShell.execute("create table src(x int,y string) partitioned by (p1 bigint, p2 string)");
final String catalogName = "hive";
try {
- hiveShell.insertInto("default", "src")
- .addRow(1, "a", 2013, "2013")
- .addRow(2, "b", 2013, "2013")
- .addRow(3, "c", 2014, "2014")
- .commit();
+ HiveTestUtils.createTextTableInserter(hiveShell, "default", "src")
+ .addRow(new Object[]{1, "a"})
+ .addRow(new Object[]{2, "b"})
+ .commit("p1=2013, p2='2013'");
+ HiveTestUtils.createTextTableInserter(hiveShell, "default", "src")
+ .addRow(new Object[]{3, "c"})
+ .commit("p1=2014, p2='2014'");
TableEnvironment tableEnv = HiveTestUtils.createTableEnv();
tableEnv.registerCatalog(catalogName, hiveCatalog);
Table table = tableEnv.sqlQuery("select p1, count(y) from hive.`default`.src group by p1");
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorTest.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorTest.java
index 9484e55..2c50bef 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorTest.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorTest.java
@@ -77,7 +77,7 @@ public class TableEnvHiveConnectorTest {
hiveShell.execute("create database db1");
hiveShell.execute("create table db1.src (x int, y int)");
hiveShell.execute("create table db1.part (x int) partitioned by (y int)");
- hiveShell.insertInto("db1", "src").addRow(1, 1).addRow(2, null).commit();
+ HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src").addRow(new Object[]{1, 1}).addRow(new Object[]{2, null}).commit();
TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
@@ -186,7 +186,7 @@ public class TableEnvHiveConnectorTest {
try {
// non-partitioned
hiveShell.execute("create table db1.dest (x int, y string)");
- hiveShell.insertInto("db1", "dest").addRow(1, "a").addRow(2, "b").commit();
+ 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"));
TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
tableEnv.sqlUpdate("insert overwrite db1.dest values (3,'c')");
@@ -195,7 +195,8 @@ public class TableEnvHiveConnectorTest {
// static partition
hiveShell.execute("create table db1.part(x int) partitioned by (y int)");
- hiveShell.insertInto("db1", "part").addRow(1, 1).addRow(2, 2).commit();
+ 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.sqlUpdate("insert overwrite db1.part partition (y=1) select 100");
tableEnv.execute("insert overwrite static partition");
@@ -217,7 +218,7 @@ public class TableEnvHiveConnectorTest {
hiveShell.execute("create database db1");
try {
hiveShell.execute("create table db1.src (x int)");
- hiveShell.insertInto("db1", "src").addRow(1).addRow(2).commit();
+ HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src").addRow(new Object[]{1}).addRow(new Object[]{2}).commit();
hiveShell.execute("create table db1.dest (x int) partitioned by (p1 string, p2 double)");
TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
tableEnv.sqlUpdate("insert into db1.dest partition (p1='1''1', p2=1.1) select x from db1.src");
@@ -234,7 +235,11 @@ public class TableEnvHiveConnectorTest {
hiveShell.execute("create database db1");
try {
hiveShell.execute("create table db1.src (x int, y string, z double)");
- hiveShell.insertInto("db1", "src").addRow(1, "a", 1.1).addRow(2, "a", 2.2).addRow(3, "b", 3.3).commit();
+ 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();
hiveShell.execute("create table db1.dest (x int) partitioned by (p1 string, p2 double)");
TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
tableEnv.sqlUpdate("insert into db1.dest select * from db1.src");
@@ -251,7 +256,7 @@ public class TableEnvHiveConnectorTest {
hiveShell.execute("create database db1");
try {
hiveShell.execute("create table db1.src (x int, y string)");
- hiveShell.insertInto("db1", "src").addRow(1, "a").addRow(2, "b").commit();
+ HiveTestUtils.createTextTableInserter(hiveShell, "db1", "src").addRow(new Object[]{1, "a"}).addRow(new Object[]{2, "b"}).commit();
hiveShell.execute("create table db1.dest (x int) partitioned by (p1 double, p2 string)");
TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
tableEnv.sqlUpdate("insert into db1.dest partition (p1=1.1) select x,y from db1.src");
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogHiveMetadataTest.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogHiveMetadataTest.java
index 0fe10d7..fb5c04a 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogHiveMetadataTest.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogHiveMetadataTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.catalog.CatalogPartitionSpec;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.CatalogTableImpl;
+import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataBase;
import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataBinary;
@@ -79,26 +80,32 @@ public class HiveCatalogHiveMetadataTest extends HiveCatalogTestBase {
// ------ table and column stats ------
@Test
public void testAlterTableColumnStatistics() throws Exception {
+ String hiveVersion = ((HiveCatalog) catalog).getHiveVersion();
+ boolean supportDateStats = hiveVersion.compareTo(HiveShimLoader.HIVE_VERSION_V1_2_0) >= 0;
catalog.createDatabase(db1, createDb(), false);
- TableSchema tableSchema = TableSchema.builder()
- .field("first", DataTypes.STRING())
- .field("second", DataTypes.INT())
- .field("third", DataTypes.BOOLEAN())
- .field("fourth", DataTypes.DATE())
- .field("fifth", DataTypes.DOUBLE())
- .field("sixth", DataTypes.BIGINT())
- .field("seventh", DataTypes.BYTES())
- .build();
+ TableSchema.Builder builder = TableSchema.builder()
+ .field("first", DataTypes.STRING())
+ .field("second", DataTypes.INT())
+ .field("third", DataTypes.BOOLEAN())
+ .field("fourth", DataTypes.DOUBLE())
+ .field("fifth", DataTypes.BIGINT())
+ .field("sixth", DataTypes.BYTES());
+ if (supportDateStats) {
+ builder.field("seventh", DataTypes.DATE());
+ }
+ TableSchema tableSchema = builder.build();
CatalogTable catalogTable = new CatalogTableImpl(tableSchema, getBatchTableProperties(), TEST_COMMENT);
catalog.createTable(path1, catalogTable, false);
Map<String, CatalogColumnStatisticsDataBase> columnStatisticsDataBaseMap = new HashMap<>();
columnStatisticsDataBaseMap.put("first", new CatalogColumnStatisticsDataString(10, 5.2, 3, 100));
columnStatisticsDataBaseMap.put("second", new CatalogColumnStatisticsDataLong(0, 1000, 3, 0));
columnStatisticsDataBaseMap.put("third", new CatalogColumnStatisticsDataBoolean(15, 20, 3));
- columnStatisticsDataBaseMap.put("fourth", new CatalogColumnStatisticsDataDate(new Date(71L), new Date(17923L), 1321, 0L));
- columnStatisticsDataBaseMap.put("fifth", new CatalogColumnStatisticsDataDouble(15.02, 20.01, 3, 10));
- columnStatisticsDataBaseMap.put("sixth", new CatalogColumnStatisticsDataLong(0, 20, 3, 2));
- columnStatisticsDataBaseMap.put("seventh", new CatalogColumnStatisticsDataBinary(150, 20, 3));
+ columnStatisticsDataBaseMap.put("fourth", new CatalogColumnStatisticsDataDouble(15.02, 20.01, 3, 10));
+ columnStatisticsDataBaseMap.put("fifth", new CatalogColumnStatisticsDataLong(0, 20, 3, 2));
+ columnStatisticsDataBaseMap.put("sixth", new CatalogColumnStatisticsDataBinary(150, 20, 3));
+ if (supportDateStats) {
+ columnStatisticsDataBaseMap.put("seventh", new CatalogColumnStatisticsDataDate(new Date(71L), new Date(17923L), 1321, 0L));
+ }
CatalogColumnStatistics catalogColumnStatistics = new CatalogColumnStatistics(columnStatisticsDataBaseMap);
catalog.alterTableColumnStatistics(path1, catalogColumnStatistics, false);
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogTestBase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogTestBase.java
index 232c250..88895a3 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogTestBase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogTestBase.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.catalog.hive;
import org.apache.flink.table.catalog.CatalogTestBase;
import org.apache.flink.table.catalog.ObjectPath;
+import org.junit.Assume;
import org.junit.Test;
/**
@@ -28,6 +29,16 @@ import org.junit.Test;
*/
public abstract class HiveCatalogTestBase extends CatalogTestBase {
+ // ------ table and column stats ------
+
+ @Override
+ @Test
+ public void testAlterTableStats() throws Exception {
+ String hiveVersion = ((HiveCatalog) catalog).getHiveVersion();
+ Assume.assumeTrue(hiveVersion.compareTo("1.2.1") >= 0);
+ super.testAlterTableStats();
+ }
+
// ------ functions ------
@Test
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 2e06d51..ec142cb 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
@@ -37,15 +37,20 @@ import org.apache.flink.types.Row;
import org.apache.flink.util.AbstractID;
import org.apache.flink.util.StringUtils;
+import com.klarna.hiverunner.HiveShell;
import org.apache.hadoop.hive.conf.HiveConf;
import org.junit.rules.TemporaryFolder;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
import java.io.IOException;
import java.net.BindException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
@@ -63,6 +68,8 @@ 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};
+
/**
* Create a HiveCatalog with an embedded Hive Metastore.
*/
@@ -138,4 +145,106 @@ public class HiveTestUtils {
ArrayList<byte[]> data = result.getAccumulatorResult(id);
return SerializedListAccumulator.deserializeList(data, serializer);
}
+
+ // 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);
+ }
+
+ /**
+ * insert table operation.
+ */
+ public static class TextTableInserter {
+
+ private final HiveShell hiveShell;
+ private final String dbName;
+ private final String tableName;
+ private final List<Object[]> rows;
+
+ public TextTableInserter(HiveShell hiveShell, String dbName, String tableName) {
+ this.hiveShell = hiveShell;
+ this.dbName = dbName;
+ this.tableName = tableName;
+ rows = new ArrayList<>();
+ }
+
+ public TextTableInserter addRow(Object[] row) {
+ rows.add(row);
+ return this;
+ }
+
+ public void commit() {
+ 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)));
+ }
+ }
+ 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);
+ }
+ }
+
+ private String toText(Object[] row) {
+ StringBuilder builder = new StringBuilder();
+ for (Object col : row) {
+ if (builder.length() > 0) {
+ builder.appendCodePoint(SEPARATORS[0]);
+ }
+ String colStr = toText(col);
+ if (colStr != null) {
+ builder.append(toText(col));
+ }
+ }
+ return builder.toString();
+ }
+
+ private String toText(Object obj) {
+ if (obj == null) {
+ return null;
+ }
+ StringBuilder builder = new StringBuilder();
+ if (obj instanceof Map) {
+ for (Object key : ((Map) obj).keySet()) {
+ if (builder.length() > 0) {
+ builder.appendCodePoint(SEPARATORS[1]);
+ }
+ builder.append(toText(key));
+ builder.appendCodePoint(SEPARATORS[2]);
+ builder.append(toText(((Map) obj).get(key)));
+ }
+ } else if (obj instanceof Object[]) {
+ Object[] array = (Object[]) obj;
+ for (Object element : array) {
+ if (builder.length() > 0) {
+ builder.appendCodePoint(SEPARATORS[1]);
+ }
+ builder.append(toText(element));
+ }
+ } else if (obj instanceof List) {
+ for (Object element : (List) obj) {
+ if (builder.length() > 0) {
+ builder.appendCodePoint(SEPARATORS[1]);
+ }
+ builder.append(toText(element));
+ }
+ } else {
+ builder.append(obj);
+ }
+ return builder.toString();
+ }
+ }
}
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/functions/hive/HiveGenericUDFTest.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/functions/hive/HiveGenericUDFTest.java
index 938f794..f7c74dc 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/functions/hive/HiveGenericUDFTest.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/functions/hive/HiveGenericUDFTest.java
@@ -34,11 +34,11 @@ import org.apache.hadoop.hive.ql.udf.generic.GenericUDFCase;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFCeil;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFCoalesce;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFDateDiff;
-import org.apache.hadoop.hive.ql.udf.generic.GenericUDFDateFormat;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFDecode;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFMapKeys;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFStringToMap;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFStruct;
+import org.junit.Assume;
import org.junit.Test;
import java.lang.reflect.InvocationTargetException;
@@ -52,6 +52,7 @@ import static org.junit.Assert.assertEquals;
*/
public class HiveGenericUDFTest {
private static HiveShim hiveShim = HiveShimLoader.loadHiveShim(HiveShimLoader.getHiveVersion());
+ private static final boolean HIVE_120_OR_LATER = HiveShimLoader.getHiveVersion().compareTo(HiveShimLoader.HIVE_VERSION_V1_2_0) >= 0;
@Test
public void testAbs() {
@@ -111,12 +112,13 @@ public class HiveGenericUDFTest {
}
@Test
- public void testDateFormat() throws NoSuchMethodException, IllegalAccessException, InvocationTargetException {
+ public void testDateFormat() throws Exception {
+ Assume.assumeTrue(HIVE_120_OR_LATER);
String constYear = "y";
String constMonth = "M";
HiveGenericUDF udf = init(
- GenericUDFDateFormat.class,
+ Class.forName("org.apache.hadoop.hive.ql.udf.generic.GenericUDFDateFormat"),
new Object[] {
null,
constYear
@@ -130,7 +132,7 @@ public class HiveGenericUDFTest {
assertEquals("2009", udf.eval("2009-08-31", constYear));
udf = init(
- GenericUDFDateFormat.class,
+ Class.forName("org.apache.hadoop.hive.ql.udf.generic.GenericUDFDateFormat"),
new Object[] {
null,
constMonth
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/functions/hive/HiveGenericUDTFTest.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/functions/hive/HiveGenericUDTFTest.java
index 12cb080..755f56f 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/functions/hive/HiveGenericUDTFTest.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/functions/hive/HiveGenericUDTFTest.java
@@ -33,6 +33,7 @@ import org.apache.hadoop.hive.ql.udf.generic.GenericUDTFInline;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTFPosExplode;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTFStack;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorConverters;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
@@ -232,8 +233,15 @@ public class HiveGenericUDTFTest {
* Test over sum int udtf.
*/
public static class TestOverSumIntUDTF extends GenericUDTF {
+
+ ObjectInspectorConverters.Converter[] converters;
+
@Override
public StructObjectInspector initialize(ObjectInspector[] argOIs) throws UDFArgumentException {
+ converters = new ObjectInspectorConverters.Converter[argOIs.length];
+ for (int i = 0; i < converters.length; i++) {
+ converters[i] = ObjectInspectorConverters.getConverter(argOIs[i], PrimitiveObjectInspectorFactory.javaIntObjectInspector);
+ }
return ObjectInspectorFactory.getStandardStructObjectInspector(
Collections.singletonList("col1"),
Collections.singletonList(PrimitiveObjectInspectorFactory.javaIntObjectInspector));
@@ -242,8 +250,8 @@ public class HiveGenericUDTFTest {
@Override
public void process(Object[] args) throws HiveException {
int total = 0;
- for (Object arg : args) {
- total += (int) arg;
+ for (int i = 0; i < args.length; i++) {
+ total += (int) converters[i].convert(args[i]);
}
for (Object ignored : args) {
forward(total);