jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

!522 improve the performance of hbase connector 

Merge pull request !522 from tushengxia/hbase-performance-improve
This commit is contained in:
i-robot 2021-01-07 09:24:07 +08:00 committed by Gitee
parent 6578db566e d4753a7d8c
commit 38ecf780bb
27 changed files with 434 additions and 396 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hetu-docs/en/connector/HBase.md
View File

@ -96,7 +96,7 @@ debug=true;
| row_id | String | The first column name | No | row_id is the column name corresponding to rowkey in the hbase table |
| hbase_table_name | String | null | No | hbase_table_name specifies the tablespace and table name on the hbase data source to be linked, use ":" connect tablespace and table name, the default tablespace is "default". |
| external | Boolean | true | No | If external is true, it means that the table is a mapping table of the table in the hbase data source. It does not support deleting the original table on the hbase data source; if external is false, the table on hbase data source will be deleted at the same time as the local hbase table is deleted. |
| split\_by\_char| String| 0~9,a~z,A~Z| No| split\_by\_char is the basis for sharding, if the first character of rowkey consists of digits, sharding can be performed based on different digits to improve concurrency. Different types of symbols are separated by commas. If this parameter is set incorrectly, the query result may be incomplete. Set it based on the actual rowkey.|
## Data Types

1
hetu-docs/zh/connector/HBase.md
View File

@ -97,6 +97,7 @@ debug=true;
| row\_id| String| 第一个列名| 否| row\_id为HBase表中RowKey对应的列名|
| hbase\_table\_name| String| NULL| 否| hbase\_table\_name指定要链接的HBase数据源上的表空间和表名，使用“:”连接表空间和表名，默认表空间为“default”。|
| external| Boolean| true| 否| 如果external为true，表示该表是HBase数据源中表的映射表。不支持删除HBase数据源上原有的表。当external为false时，删除本地HBase表的同时也会删除HBase数据源上的表。|
| split\_by\_char| String| 0~9,a~z,A~Z| 否| split\_by\_char为分片切割的依据，若RowKey的第一个字符由数字构成，则可以根据不同的数字进行分片切割，提高查询并发度。不同类型的符号用逗号隔开。如果设置不当，会导致查询数据结果不完整，请根据RowKey的实际情况进行配置。|
## 数据说明

168
hetu-hbase/pom.xml
View File

@ -37,29 +37,6 @@
</exclusions>
</dependency>
<dependency>
<groupId>org.weakref</groupId>
<artifactId>jmxutils</artifactId>
<version>${dep.jmxutils.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>io.prestosql.hadoop</groupId>
<artifactId>hadoop-apache</artifactId>
<version>${dep.prestosql.apache.hadoop.version}</version>
<exclusions>
<exclusion>
<groupId>org.antlr</groupId>
<artifactId>antlr4-runtime</artifactId>
</exclusion>
<exclusion>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
@ -71,6 +48,11 @@
<artifactId>hetu-common</artifactId>
</dependency>
<dependency>
<groupId>io.prestosql.hadoop</groupId>
<artifactId>hadoop-apache</artifactId>
</dependency>
<!-- Presto SPI -->
<dependency>
<groupId>io.hetu.core</groupId>
@ -149,127 +131,23 @@
<version>${version.commons-lang3}</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-common</artifactId>
<version>${version.hbase}</version>
<exclusions>
<exclusion>
<artifactId>commons-codec</artifactId>
<groupId>commons-codec</groupId>
</exclusion>
<exclusion>
<artifactId>jackson-databind</artifactId>
<groupId>com.fasterxml.jackson.core</groupId>
</exclusion>
<exclusion>
<artifactId>hadoop-common</artifactId>
<groupId>org.apache.hadoop</groupId>
</exclusion>
<exclusion>
<artifactId>guice-servlet</artifactId>
<groupId>com.google.inject.extensions</groupId>
</exclusion>
<exclusion>
<artifactId>jackson-annotations</artifactId>
<groupId>com.fasterxml.jackson.core</groupId>
</exclusion>
<exclusion>
<artifactId>commons-logging</artifactId>
<groupId>commons-logging</groupId>
</exclusion>
<exclusion>
<artifactId>jaxb-api</artifactId>
<groupId>javax.xml.bind</groupId>
</exclusion>
<exclusion>
<artifactId>jackson-jaxrs-json-provider</artifactId>
<groupId>com.fasterxml.jackson.jaxrs</groupId>
</exclusion>
<exclusion>
<artifactId>jackson-module-jaxb-annotations</artifactId>
<groupId>com.fasterxml.jackson.module</groupId>
</exclusion>
<exclusion>
<artifactId>jackson-core</artifactId>
<groupId>com.fasterxml.jackson.core</groupId>
</exclusion>
<exclusion>
<artifactId>hadoop-yarn-common</artifactId>
<groupId>org.apache.hadoop</groupId>
</exclusion>
<exclusion>
<artifactId>junit</artifactId>
<groupId>junit</groupId>
</exclusion>
<exclusion>
<artifactId>protobuf-java</artifactId>
<groupId>com.google.protobuf</groupId>
</exclusion>
<exclusion>
<artifactId>hadoop-annotations</artifactId>
<groupId>org.apache.hadoop</groupId>
</exclusion>
<exclusion>
<artifactId>jetty-util</artifactId>
<groupId>org.mortbay.jetty</groupId>
</exclusion>
<exclusion>
<artifactId>hadoop-mapreduce-client-core</artifactId>
<groupId>org.apache.hadoop</groupId>
</exclusion>
<exclusion>
<artifactId>error_prone_annotations</artifactId>
<groupId>com.google.errorprone</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>${version.hbase}</version>
<exclusions>
<exclusion>
<artifactId>commons-codec</artifactId>
<groupId>commons-codec</groupId>
</exclusion>
<exclusion>
<artifactId>htrace-core4</artifactId>
<groupId>org.apache.htrace</groupId>
<artifactId>error_prone_annotations</artifactId>
<groupId>com.google.errorprone</groupId>
</exclusion>
<exclusion>
<artifactId>jcodings</artifactId>
<groupId>org.jruby.jcodings</groupId>
</exclusion>
<exclusion>
<artifactId>jackson-databind</artifactId>
<groupId>com.fasterxml.jackson.core</groupId>
</exclusion>
<exclusion>
<artifactId>hbase-common</artifactId>
<groupId>org.apache.hbase</groupId>
</exclusion>
<exclusion>
<artifactId>commons-logging</artifactId>
<groupId>commons-logging</groupId>
</exclusion>
<exclusion>
<artifactId>junit</artifactId>
<groupId>junit</groupId>
</exclusion>
<exclusion>
<artifactId>hadoop-common</artifactId>
<groupId>org.apache.hadoop</groupId>
</exclusion>
<exclusion>
<artifactId>netty-all</artifactId>
<groupId>io.netty</groupId>
</exclusion>
<exclusion>
<artifactId>hadoop-mapreduce-client-core</artifactId>
<groupId>org.apache.hadoop</groupId>
</exclusion>
<exclusion>
<artifactId>hadoop-auth</artifactId>
<groupId>org.apache.hadoop</groupId>
@ -287,6 +165,19 @@
<groupId>com.fasterxml.jackson.core</groupId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-common</artifactId>
<version>${version.hbase}</version>
<exclusions>
<exclusion>
<artifactId>hadoop-common</artifactId>
<groupId>org.apache.hadoop</groupId>
</exclusion>
</exclusions>
</dependency>
<!-- for testing -->
<dependency>
<groupId>org.testng</groupId>
@ -294,10 +185,27 @@
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.weakref</groupId>
<artifactId>jmxutils</artifactId>
<version>${dep.jmxutils.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>io.hetu.core</groupId>
<artifactId>presto-tests</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<artifactId>plexus-cipher</artifactId>
<groupId>org.sonatype.plexus</groupId>
</exclusion>
<exclusion>
<artifactId>plexus-classworlds</artifactId>
<groupId>org.codehaus.plexus</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
@ -353,4 +261,4 @@
</plugins>
</build>
</project>
</project>

86
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/conf/HBaseTableProperties.java
View File

@ -78,6 +78,10 @@ public final class HBaseTableProperties
* hbase_table_name
*/
public static final String HBASE_TABLE_NAME = "hbase_table_name";
/**
* split_by_char
*/
public static final String SPLIT_BY_CHAR = "split_by_char";
private static final String COLON_SPLITTER = ":";
private static final String COMMA_SPLITTER = ",";
private static final String PIPE_SPLITTER = "\\|";
@ -146,8 +150,14 @@ public final class HBaseTableProperties
"Point to the table of hbase, establish the mapping of external access",
null,
false);
PropertyMetadata<String> s8 =
stringProperty(
SPLIT_BY_CHAR,
"Create multi-splits by the first char of rowKey",
"0~9,a~z,A~Z",
false);
tableProperties = ImmutableList.of(s1, s2, s3, s4, s5, s6, s7);
tableProperties = ImmutableList.of(s1, s2, s3, s4, s5, s6, s7, s8);
}
public List<PropertyMetadata<?>> getTableProperties()
@ -165,16 +175,14 @@ public final class HBaseTableProperties
*/
public static Optional<Map<String, Pair<String, String>>> getColumnMapping(Map<String, Object> tableProperties)
{
requireNonNull(tableProperties);
String strMapping = String.valueOf(tableProperties.get(COLUMN_MAPPING));
if (Constants.S_NULL.equals(strMapping)) {
Optional<String> strMapping = getStringFromProperties(COLUMN_MAPPING, tableProperties);
if (!strMapping.isPresent()) {
return Optional.empty();
}
// Parse out the column mapping list : "column_name:family:qualifier"
ImmutableMap.Builder<String, Pair<String, String>> mapping = ImmutableMap.builder();
for (String m1 : strMapping.split(COMMA_SPLITTER)) {
for (String m1 : strMapping.get().split(COMMA_SPLITTER)) {
String[] tokens = m1.trim().split(COLON_SPLITTER);
checkState(
tokens.length == Constants.NUMBER3,
@ -194,14 +202,13 @@ public final class HBaseTableProperties
*/
public static Optional<List<String>> getIndexColumns(Map<String, Object> tableProperties)
{
requireNonNull(tableProperties);
String indexColumns = String.valueOf(tableProperties.get(INDEX_COLUMNS));
if (Constants.S_NULL.equals(indexColumns)) {
Optional<String> indexColumns = getStringFromProperties(INDEX_COLUMNS, tableProperties);
if (!indexColumns.isPresent()) {
return Optional.empty();
}
return Optional.of(Arrays.asList(StringUtils.split(indexColumns, ',')));
else {
return Optional.of(Arrays.asList(StringUtils.split(indexColumns.get(), ',')));
}
}
/**
@ -212,14 +219,7 @@ public final class HBaseTableProperties
*/
public static Optional<String> getIndexColumnsAsStr(Map<String, Object> tableProperties)
{
requireNonNull(tableProperties);
String indexColumns = String.valueOf(tableProperties.get(INDEX_COLUMNS));
if (Constants.S_NULL.equals(indexColumns)) {
return Optional.empty();
}
return Optional.of(indexColumns);
return getStringFromProperties(INDEX_COLUMNS, tableProperties);
}
/**
@ -230,17 +230,15 @@ public final class HBaseTableProperties
*/
public static Optional<Map<String, Set<String>>> getLocalityGroups(Map<String, Object> tableProperties)
{
requireNonNull(tableProperties);
String groupStr = String.valueOf(tableProperties.get(LOCALITY_GROUPS));
if (Constants.S_NULL.equals(groupStr)) {
Optional<String> groupStr = getStringFromProperties(LOCALITY_GROUPS, tableProperties);
if (!groupStr.isPresent()) {
return Optional.empty();
}
ImmutableMap.Builder<String, Set<String>> groups = ImmutableMap.builder();
// Split all configured locality groups
for (String group : groupStr.split(PIPE_SPLITTER)) {
for (String group : groupStr.get().split(PIPE_SPLITTER)) {
String[] locGroups = group.trim().split(COLON_SPLITTER);
if (locGroups.length != Constants.NUMBER2) {
throw new PrestoException(
@ -269,13 +267,7 @@ public final class HBaseTableProperties
*/
public static Optional<String> getRowId(Map<String, Object> tableProperties)
{
requireNonNull(tableProperties);
String rowId = String.valueOf(tableProperties.get(ROW_ID));
if (Constants.S_NULL.equals(rowId)) {
return Optional.empty();
}
return Optional.ofNullable(rowId);
return getStringFromProperties(ROW_ID, tableProperties);
}
/**
@ -286,13 +278,7 @@ public final class HBaseTableProperties
*/
public static Optional<String> getSerializerClass(Map<String, Object> tableProperties)
{
requireNonNull(tableProperties);
String serializerClass = String.valueOf(tableProperties.get(SERIALIZER));
if (Constants.S_NULL.equals(serializerClass)) {
return Optional.empty();
}
return Optional.ofNullable(serializerClass);
return getStringFromProperties(SERIALIZER, tableProperties);
}
/**
@ -316,13 +302,29 @@ public final class HBaseTableProperties
* @return Optional<String>
*/
public static Optional<String> getHBaseTableName(Map<String, Object> tableProperties)
{
return getStringFromProperties(HBASE_TABLE_NAME, tableProperties);
}
/**
* getSplitByChar
*
* @param tableProperties tableProperties
* @return Optional<String>
*/
public static Optional<String> getSplitByChar(Map<String, Object> tableProperties)
{
return getStringFromProperties(SPLIT_BY_CHAR, tableProperties);
}
private static Optional<String> getStringFromProperties(String key, Map<String, Object> tableProperties)
{
requireNonNull(tableProperties);
String hbaseTableName = String.valueOf(tableProperties.get(HBASE_TABLE_NAME));
if (Constants.S_NULL.equals(hbaseTableName)) {
String value = String.valueOf(tableProperties.get(key));
if (Constants.S_NULL.equals(value)) {
return Optional.empty();
}
return Optional.ofNullable(hbaseTableName);
return Optional.ofNullable(value);
}
}

11
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/connector/HBaseConnection.java
View File

@ -542,7 +542,8 @@ public class HBaseConnection
HBaseTableProperties.isExternal(tableProperties),
HBaseTableProperties.getSerializerClass(tableProperties),
HBaseTableProperties.getIndexColumnsAsStr(meta.getProperties()),
HBaseTableProperties.getHBaseTableName(meta.getProperties()));
HBaseTableProperties.getHBaseTableName(meta.getProperties()),
HBaseTableProperties.getSplitByChar(meta.getProperties()));
table.setColumnsToMap(columnHandleMap);
@ -564,7 +565,9 @@ public class HBaseConnection
}
}
else {
if (table.isExternal()) {
String hbaseTableName = table.getFullTableName().replace(Constants.POINT, Constants.SEPARATOR);
table.setHbaseTableName(Optional.of(hbaseTableName));
if (table.isExternal() && !existTable(hbaseTableName)) {
throw new PrestoException(
HBaseErrorCode.HBASE_CREATE_ERROR,
format("Use lk creating new HBase table [%s], we must specify 'with(external=false)'. ", table.getTable()));
@ -572,12 +575,8 @@ public class HBaseConnection
// create namespace if not exist
createNamespaceIfNotExist(this.getHbaseAdmin(), table.getSchema());
String hbaseTableName = table.getFullTableName().replace(Constants.POINT, Constants.SEPARATOR);
table.setHbaseTableName(Optional.ofNullable(hbaseTableName));
// create hbase table
createHBaseTable(table);
// save tableCatalog to memory and file
hBaseMetastore.addHBaseTable(table);
return table;

5
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/connector/HBaseConnectorFactory.java
View File

@ -75,12 +75,13 @@ public class HBaseConnectorFactory
public Connector create(String catalogName, Map<String, String> config, ConnectorContext context)
{
requireNonNull(config, "config is null");
HBaseConnectorId.setConnectorId(catalogName);
try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
hetuMetastore = context.getHetuMetastore();
Bootstrap app = new Bootstrap(new HBaseModule(), this.module);
Bootstrap app = new Bootstrap(binder ->
binder.bind(HetuMetastore.class).toInstance(context.getHetuMetastore()),
new HBaseModule(), this.module);
Injector injector =
app.strictConfig().doNotInitializeLogging().quiet().setRequiredConfigurationProperties(config).initialize();
return injector.getInstance(HBaseConnector.class);

30
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/connector/HBaseTableHandle.java
View File

@ -29,6 +29,7 @@ import io.prestosql.spi.predicate.TupleDomain;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.OptionalLong;
import static com.google.common.base.MoreObjects.toStringHelper;
import static java.util.Objects.requireNonNull;
@ -51,6 +52,7 @@ public class HBaseTableHandle
private final TupleDomain<ColumnHandle> constraint;
private final List<HBaseColumnHandle> columns;
private final int rowIdOrdinal;
private final OptionalLong limit;
/**
* constructor
@ -65,6 +67,7 @@ public class HBaseTableHandle
* @param constraint constraint
* @param columns columns
* @param rowIdOrdinal rowIdOrdinal
* @param limit limit
*/
@JsonCreator
public HBaseTableHandle(
@ -77,7 +80,8 @@ public class HBaseTableHandle
@JsonProperty("indexColumns") String indexColumns,
@JsonProperty("constraint") TupleDomain<ColumnHandle> constraint,
@JsonProperty("columns") List<HBaseColumnHandle> columns,
@JsonProperty("rowIdOrdinal") int rowIdOrdinal)
@JsonProperty("rowIdOrdinal") int rowIdOrdinal,
@JsonProperty("limit") OptionalLong limit)
{
this.external = external;
this.rowId = requireNonNull(rowId, "rowId is null");
@ -89,6 +93,7 @@ public class HBaseTableHandle
this.constraint = constraint;
this.columns = columns;
this.rowIdOrdinal = rowIdOrdinal;
this.limit = requireNonNull(limit, "limit is null");
}
/**
@ -100,6 +105,7 @@ public class HBaseTableHandle
* @param columns columns
* @param serializerClassName serializerClassName
* @param hbaseTableName hbaseTableName
* @param limit limit
*/
public HBaseTableHandle(
String schema,
@ -107,7 +113,8 @@ public class HBaseTableHandle
int rowIdOrdinal,
List<HBaseColumnHandle> columns,
String serializerClassName,
Optional<String> hbaseTableName)
Optional<String> hbaseTableName,
OptionalLong limit)
{
this(
schema,
@ -119,7 +126,8 @@ public class HBaseTableHandle
"",
TupleDomain.all(),
columns,
rowIdOrdinal);
rowIdOrdinal,
limit);
}
/**
@ -132,6 +140,7 @@ public class HBaseTableHandle
* @param serializerClassName serializerClassName
* @param hbaseTableName hbaseTableName
* @param indexColumns indexColumns
* @param limit limit
*/
public HBaseTableHandle(
String schema,
@ -140,7 +149,8 @@ public class HBaseTableHandle
boolean external,
String serializerClassName,
Optional<String> hbaseTableName,
String indexColumns)
String indexColumns,
OptionalLong limit)
{
this(
schema,
@ -152,7 +162,8 @@ public class HBaseTableHandle
indexColumns,
TupleDomain.all(),
ImmutableList.of(),
0);
0,
limit);
}
@JsonProperty
@ -215,6 +226,12 @@ public class HBaseTableHandle
return indexColumns;
}
@JsonProperty
public OptionalLong getLimit()
{
return limit;
}
/**
* toSchemaTableName
*
@ -262,7 +279,8 @@ public class HBaseTableHandle
oldHBaseConnectorTableHandle.getIndexColumns(),
oldHBaseConnectorTableHandle.getConstraint(),
oldHBaseConnectorTableHandle.getColumns(),
oldHBaseConnectorTableHandle.getRowIdOrdinal());
oldHBaseConnectorTableHandle.getRowIdOrdinal(),
oldHBaseConnectorTableHandle.getLimit());
}
@Override

59
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/metadata/HBaseConnectorMetadata.java
View File

@ -20,12 +20,9 @@ import com.google.common.collect.ImmutableSet;
import com.google.inject.Inject;
import io.airlift.log.Logger;
import io.airlift.slice.Slice;
import io.hetu.core.plugin.hbase.conf.HBaseTableProperties;
import io.hetu.core.plugin.hbase.connector.HBaseColumnHandle;
import io.hetu.core.plugin.hbase.connector.HBaseConnection;
import io.hetu.core.plugin.hbase.connector.HBaseTableHandle;
import io.hetu.core.plugin.hbase.utils.HBaseErrorCode;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.connector.ColumnHandle;
import io.prestosql.spi.connector.ColumnMetadata;
import io.prestosql.spi.connector.ConnectorInsertTableHandle;
@ -40,6 +37,7 @@ import io.prestosql.spi.connector.ConnectorTableMetadata;
import io.prestosql.spi.connector.ConnectorTableProperties;
import io.prestosql.spi.connector.Constraint;
import io.prestosql.spi.connector.ConstraintApplicationResult;
import io.prestosql.spi.connector.LimitApplicationResult;
import io.prestosql.spi.connector.SchemaTableName;
import io.prestosql.spi.connector.SchemaTablePrefix;
import io.prestosql.spi.connector.TableNotFoundException;
@ -58,7 +56,6 @@ import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import static com.google.common.base.Preconditions.checkState;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
/**
@ -112,7 +109,8 @@ public class HBaseConnectorMetadata
table.isExternal(),
table.getSerializerClassName(),
table.getHbaseTableName(),
table.getIndexColumns());
table.getIndexColumns(),
OptionalLong.empty());
}
@Override
@ -168,7 +166,7 @@ public class HBaseConnectorMetadata
return Optional.ofNullable(tableMetadata).orElse(tableMetadata);
}
return new ConnectorTableMetadata(tableName, table.getColumnMetadatas());
return new ConnectorTableMetadata(tableName, table.getColumnMetadatas(), table.getTableProperties());
}
@Override
@ -177,12 +175,6 @@ public class HBaseConnectorMetadata
{
checkNoRollback();
SchemaTableName tableName = tableMetadata.getTable();
if (HBaseTableProperties.isExternal(tableMetadata.getProperties())) {
throw new PrestoException(
HBaseErrorCode.HBASE_CREATE_ERROR,
format("Use lk creating new HBase table [%s], we must specify 'with(external=false)'. ", tableName.toString()));
}
HBaseTable table = hbaseConn.createTable(tableMetadata);
// support create table xxx as select * from yyy
HBaseTableHandle handle =
@ -196,7 +188,8 @@ public class HBaseConnectorMetadata
table.getIndexColumns(),
TupleDomain.all(),
table.getColumns(),
table.getRowIdOrdinal());
table.getRowIdOrdinal(),
OptionalLong.empty());
setRollback(() -> rollbackCreateTable(table));
return handle;
@ -227,7 +220,8 @@ public class HBaseConnectorMetadata
table.getRowIdOrdinal(),
table.getColumns(),
table.getSerializerClassName(),
table.getHbaseTableName());
table.getHbaseTableName(),
OptionalLong.empty());
return handle;
}
@ -406,10 +400,45 @@ public class HBaseConnectorMetadata
tableHandle.getFullTableName(),
newDomain,
tableHandle.getColumns(),
tableHandle.getRowIdOrdinal());
tableHandle.getRowIdOrdinal(),
tableHandle.getLimit());
return Optional.of(new ConstraintApplicationResult<>(newTableHandle, constraint.getSummary()));
}
@Override
public Optional<LimitApplicationResult<ConnectorTableHandle>> applyLimit(
ConnectorSession session, ConnectorTableHandle connectorTableHandle, long limit)
{
HBaseTableHandle tableHandle;
if (connectorTableHandle instanceof HBaseTableHandle) {
tableHandle = (HBaseTableHandle) connectorTableHandle;
}
else {
return Optional.empty();
}
if (tableHandle.getLimit().isPresent() && tableHandle.getLimit().getAsLong() <= limit) {
return Optional.empty();
}
HBaseTableHandle newTableHandle =
new HBaseTableHandle(
tableHandle.getSchema(),
tableHandle.getTable(),
tableHandle.getRowId(),
tableHandle.isExternal(),
tableHandle.getSerializerClassName(),
tableHandle.getHbaseTableName(),
tableHandle.getFullTableName(),
tableHandle.getConstraint(),
tableHandle.getColumns(),
tableHandle.getRowIdOrdinal(),
OptionalLong.of(limit));
return Optional.of(new LimitApplicationResult<>(newTableHandle, false));
}
@Override
public ColumnHandle getUpdateRowIdColumnHandle(ConnectorSession session, ConnectorTableHandle tableHandle)
{

39
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/metadata/HBaseTable.java
View File

@ -18,6 +18,7 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import io.airlift.log.Logger;
import io.hetu.core.plugin.hbase.conf.HBaseTableProperties;
import io.hetu.core.plugin.hbase.connector.HBaseColumnHandle;
import io.hetu.core.plugin.hbase.utils.Constants;
import io.prestosql.spi.connector.ColumnHandle;
@ -28,6 +29,7 @@ import org.codehaus.jettison.json.JSONException;
import org.codehaus.jettison.json.JSONObject;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@ -55,6 +57,7 @@ public class HBaseTable
private final SchemaTableName schemaTableName;
private String indexColumns;
private Optional<String> hbaseTableName;
private Optional<String> splitByChar;
private Map<String, ColumnHandle> columnsMap;
/**
@ -68,6 +71,7 @@ public class HBaseTable
* @param serializerClassName serializerClassName
* @param indexColumns indexColumns
* @param hbaseTableName hbaseTableName
* @param splitByChar splitByChar
*/
@JsonCreator
public HBaseTable(
@ -78,7 +82,8 @@ public class HBaseTable
@JsonProperty("external") boolean external,
@JsonProperty("serializerClassName") Optional<String> serializerClassName,
@JsonProperty("indexColumns") Optional<String> indexColumns,
@JsonProperty("hbaseTableName") Optional<String> hbaseTableName)
@JsonProperty("hbaseTableName") Optional<String> hbaseTableName,
@JsonProperty("splitByChar") Optional<String> splitByChar)
{
this.external = external;
this.rowId = requireNonNull(rowId, "rowId is null");
@ -88,6 +93,7 @@ public class HBaseTable
this.serializerClassName = serializerClassName.orElse("");
this.indexColumns = indexColumns.orElse("");
this.hbaseTableName = hbaseTableName;
this.splitByChar = splitByChar;
boolean flag = false;
int rowIdOrdinal0 = 0;
@ -175,6 +181,35 @@ public class HBaseTable
return this.columnsMap;
}
/**
* getSplitByChar
*
* @return all chars range
*/
@JsonProperty
public Optional<String> getSplitByChar()
{
return splitByChar;
}
/**
* get table properties
*
* @return all table properties
*/
@JsonProperty
public Map<String, Object> getTableProperties()
{
Map<String, Object> properties = new HashMap<>();
properties.put(HBaseTableProperties.EXTERNAL, isExternal());
properties.put(HBaseTableProperties.INDEX_COLUMNS, getIndexColumns());
properties.put(HBaseTableProperties.SPLIT_BY_CHAR, getSplitByChar().orElse(""));
properties.put(HBaseTableProperties.HBASE_TABLE_NAME, getHbaseTableName().orElse(""));
properties.put(HBaseTableProperties.ROW_ID, getRowId());
properties.put(HBaseTableProperties.SERIALIZER, getSerializerClassName());
return properties;
}
/**
* setHbaseTableName
*
@ -290,6 +325,7 @@ public class HBaseTable
.add("external", external)
.add("serializerClassName", serializerClassName)
.add("hbaseTableName", hbaseTableName)
.add("splitByChar", splitByChar)
.toString();
}
@ -312,6 +348,7 @@ public class HBaseTable
jo.put("table", table);
jo.put("indexColumns", indexColumns);
jo.put("hbaseTableName", hbaseTableName.orElse(""));
jo.put("splitByChar", splitByChar.orElse(""));
JSONArray jac = new JSONArray();
JSONObject joc;

4
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/metadata/HetuHBaseMetastore.java
View File

@ -204,7 +204,8 @@ public class HetuHBaseMetastore
Constants.S_TRUE.equals(tableParam.get(Constants.S_EXTERNAL).toLowerCase(Locale.ENGLISH)),
Optional.ofNullable(tableParam.get(Constants.S_SERIALIZER_CLASS_NAME)),
Optional.ofNullable(tableParam.get(Constants.S_INDEX_COLUMNS)),
Optional.ofNullable(tableParam.get(Constants.S_HBASE_TABLE_NAME)));
Optional.ofNullable(tableParam.get(Constants.S_HBASE_TABLE_NAME)),
Optional.ofNullable(tableParam.get(Constants.S_SPLIT_BY_CHAR)));
hBaseTable.setColumnsToMap(columnsMap);
return hBaseTable;
@ -242,6 +243,7 @@ public class HetuHBaseMetastore
tableParam.put(Constants.S_ROWID, hBaseTable.getRowId());
tableParam.put(Constants.S_INDEX_COLUMNS, hBaseTable.getIndexColumns());
tableParam.put(Constants.S_HBASE_TABLE_NAME, hBaseTable.getHbaseTableName().orElse(null));
tableParam.put(Constants.S_SPLIT_BY_CHAR, hBaseTable.getSplitByChar().orElse(null));
tableEntity.setParameters(tableParam);

19
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/query/HBasePageSink.java
View File

@ -20,6 +20,7 @@ import io.airlift.slice.Slice;
import io.hetu.core.plugin.hbase.connector.HBaseColumnHandle;
import io.hetu.core.plugin.hbase.connector.HBaseConnection;
import io.hetu.core.plugin.hbase.connector.HBaseTableHandle;
import io.hetu.core.plugin.hbase.utils.Constants;
import io.hetu.core.plugin.hbase.utils.HBaseErrorCode;
import io.hetu.core.plugin.hbase.utils.serializers.HBaseRowSerializer;
import io.prestosql.spi.Page;
@ -86,13 +87,19 @@ public class HBasePageSink
// For each position within the page
List<Put> puts = new ArrayList<>();
for (int position = 0; position < page.getPositionCount(); ++position) {
// Convert Page to a Put, writing and indexing it
Put put = pageToPut(page, position);
puts.add(put);
}
try {
hbaseConn.getConn().getTable(TableName.valueOf(tablename)).put(puts);
for (int position = 0; position < page.getPositionCount(); ++position) {
// Convert Page to a Put, writing and indexing it
Put put = pageToPut(page, position);
puts.add(put);
if (puts.size() >= Constants.PUT_BATCH_SIZE) {
hbaseConn.getConn().getTable(TableName.valueOf(tablename)).put(puts);
puts.clear();
}
}
if (!puts.isEmpty()) {
hbaseConn.getConn().getTable(TableName.valueOf(tablename)).put(puts);
}
}
catch (IOException e) {
LOG.error("appendPage PUT rejected by server... cause by %s", e.getMessage());

17
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/query/HBaseRecordCursor.java
View File

@ -26,7 +26,6 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.List;
@ -187,22 +186,6 @@ public class HBaseRecordCursor
if (iterator.hasNext()) {
serializer.reset();
Result row = iterator.next();
bytesRead += row.getRow().length;
byte[] bytes;
for (HBaseColumnHandle hc : columnHandles) {
if (!hc.getName().equals(rowIdName)) {
bytes =
row.getValue(
hc.getFamily().get().getBytes(Charset.forName("UTF-8")),
hc.getQualifier().get().getBytes(Charset.forName("UTF-8")));
if (bytes != null) {
bytesRead += bytes.length;
}
}
}
serializer.deserialize(row, this.defaultValue);
return true;
}

30
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/query/HBaseRecordSet.java
View File

@ -107,7 +107,6 @@ public class HBaseRecordSet
List<HBaseColumnHandle> columnHandles)
{
requireNonNull(session, "session is null");
rowIdName = table.getRowId();
this.split = split;
this.conn = hbaseConn;
@ -135,6 +134,10 @@ public class HBaseRecordSet
this.serializer.setMapping(hc.getName(), hc.getFamily().get(), hc.getQualifier().get());
}
}
if (table.getLimit().isPresent() && table.getLimit().getAsLong() <= Integer.MAX_VALUE) {
scan.setLimit((int) table.getLimit().getAsLong());
}
LOG.info("Worker handle split：" + split.toString());
}
@Override
@ -177,14 +180,16 @@ public class HBaseRecordSet
if (filters.getFilters().size() != 0) {
scan.setFilter(filters);
}
if (split.getStartRow() != null && !split.getStartRow().isEmpty()) {
scan.setStartRow(Bytes.toBytes(split.getStartRow()));
scan.withStartRow(Bytes.toBytes(split.getStartRow()));
}
if (split.getEndRow() != null && !split.getEndRow().isEmpty()) {
scan.withStopRow(Bytes.toBytes(split.getEndRow()));
}
if (split.getEndRow() != null && !split.getEndRow().isEmpty()) {
scan.setStopRow(Bytes.toBytes(split.getEndRow()));
}
scan.setCaching(Constants.SCAN_CACHING_SIZE);
scan.setLoadColumnFamiliesOnDemand(true);
scan.setCacheBlocks(true);
scanner = conn.getConn().getTable(TableName.valueOf(table.getHbaseTableName().get())).getScanner(scan);
}
catch (IOException e) {
@ -225,11 +230,10 @@ public class HBaseRecordSet
{
FilterList andFilters = new FilterList(FilterList.Operator.MUST_PASS_ALL);
// select count(rowKey) / rowKey from table_xxx;
if (this.columnHandles.size() == 1
&& this.columnHandles.get(0).getColumnName().equals(this.split.getRowKeyName())) {
scan.setCaching(Constants.SCAN_CACHING_SIZE);
scan.setCacheBlocks(false);
// select count(*) / count(rowKey) / rowKey from table_xxx;
if ((this.columnHandles.size() == 1
&& this.columnHandles.get(0).getColumnName().equals(this.split.getRowKeyName()))
|| this.columnHandles.size() == 0) {
andFilters.addFilter(new FirstKeyOnlyFilter());
andFilters.addFilter(new KeyOnlyFilter());
}
@ -398,11 +402,13 @@ public class HBaseRecordSet
return new RowFilter(operator, new BinaryComparator(values));
}
else {
return new SingleColumnValueFilter(
SingleColumnValueFilter filter = new SingleColumnValueFilter(
Bytes.toBytes(columnHandle.getFamily().get()),
Bytes.toBytes(columnHandle.getQualifier().get()),
operator,
values);
filter.setFilterIfMissing(true);
return filter;
}
}

25
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/split/HBaseSplit.java
View File

@ -20,7 +20,6 @@ import io.hetu.core.plugin.hbase.connector.HBaseTableHandle;
import io.prestosql.spi.HostAddress;
import io.prestosql.spi.connector.ConnectorSplit;
import io.prestosql.spi.predicate.Range;
import org.apache.hadoop.hbase.HRegionInfo;
import java.util.List;
import java.util.Map;
@ -43,8 +42,6 @@ public class HBaseSplit
private final Map<Integer, List<Range>> ranges;
private final HRegionInfo regionInfo;
private final HBaseTableHandle tableHandle;
private final boolean randomSplit;
@ -58,7 +55,6 @@ public class HBaseSplit
* @param startRow startRow
* @param endRow endRow
* @param ranges search ranges
* @param regionInfo regionInfo
* @param randomSplit randomSplit
*/
@JsonCreator
@ -69,7 +65,6 @@ public class HBaseSplit
@JsonProperty("startRow") String startRow,
@JsonProperty("endRow") String endRow,
@JsonProperty("ranges") Map<Integer, List<Range>> ranges,
@JsonProperty("regionInfo") HRegionInfo regionInfo,
@JsonProperty("randomSplit") boolean randomSplit)
{
this.rowKeyName = rowKeyName;
@ -78,7 +73,6 @@ public class HBaseSplit
this.startRow = startRow;
this.endRow = endRow;
this.ranges = ranges;
this.regionInfo = regionInfo;
this.randomSplit = randomSplit;
}
@ -124,12 +118,6 @@ public class HBaseSplit
return ranges;
}
@JsonProperty
public HRegionInfo getRegionInfo()
{
return regionInfo;
}
@JsonProperty
public HBaseTableHandle getTableHandle()
{
@ -141,4 +129,17 @@ public class HBaseSplit
{
return randomSplit;
}
@Override
public String toString()
{
return "HBaseSplit{" +
"addresses='" + addresses + '\'' +
", rowKeyName='" + rowKeyName + '\'' +
", startRow='" + startRow + '\'' +
", endRow='" + endRow + '\'' +
", ranges='" + ranges + '\'' +
", randomSplit='" + randomSplit + '\'' +
'}';
}
}

184
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/split/HBaseSplitManager.java
View File

@ -20,10 +20,8 @@ import io.hetu.core.plugin.hbase.connector.HBaseColumnHandle;
import io.hetu.core.plugin.hbase.connector.HBaseConnection;
import io.hetu.core.plugin.hbase.connector.HBaseTableHandle;
import io.hetu.core.plugin.hbase.utils.Constants;
import io.hetu.core.plugin.hbase.utils.HBaseErrorCode;
import io.hetu.core.plugin.hbase.utils.Utils;
import io.prestosql.spi.HostAddress;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.connector.ColumnHandle;
import io.prestosql.spi.connector.ConnectorSession;
import io.prestosql.spi.connector.ConnectorSplitManager;
@ -35,18 +33,14 @@ import io.prestosql.spi.predicate.Domain;
import io.prestosql.spi.predicate.Range;
import io.prestosql.spi.predicate.TupleDomain;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.TableNotFoundException;
import org.apache.hadoop.hbase.client.RegionLocator;
import org.apache.hadoop.hbase.util.Pair;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static java.lang.String.format;
import java.util.stream.Collectors;
/**
* HBaseSplitManager
@ -58,12 +52,12 @@ public class HBaseSplitManager
{
private static final Logger LOG = Logger.get(HBaseSplitManager.class);
private final HBaseConnection hBaseConnection;
private final HBaseConnection hbaseConnection;
@Inject
public HBaseSplitManager(HBaseConnection hBaseConnection)
public HBaseSplitManager(HBaseConnection hbaseConnection)
{
this.hBaseConnection = hBaseConnection;
this.hbaseConnection = hbaseConnection;
}
@Override
@ -89,59 +83,100 @@ public class HBaseSplitManager
return new FixedSplitSource(splits);
}
/**
* Get splits by slicing the rowKeys, according to the first character of rowKey (user can specify it when create
* table, the default value is "0~9,a~z,A~Z"), generate many startAndEndKey pairs.
*
* @param tupleDomain tupleDomain
* @param tableHandle tableHandle
* @return splits
*/
private List<HBaseSplit> getSplitsForScan(TupleDomain<ColumnHandle> tupleDomain, HBaseTableHandle tableHandle)
{
List<HBaseSplit> splits = new ArrayList<>();
Pair<byte[][], byte[][]> startEndKeys = null;
TableName hbaseTableName = TableName.valueOf(tableHandle.getHbaseTableName().get());
try {
if (hBaseConnection.getHbaseAdmin().getTableDescriptor(hbaseTableName) != null) {
RegionLocator regionLocator =
hBaseConnection.getConn().getRegionLocator(hbaseTableName);
startEndKeys = regionLocator.getStartEndKeys();
}
}
catch (TableNotFoundException e) {
throw new PrestoException(
HBaseErrorCode.HBASE_TABLE_DNE,
format(
"table %s not found, maybe deleted by other user", tableHandle.getHbaseTableName().get()));
}
catch (IOException e) {
LOG.error(e.getMessage());
}
Map<Integer, List<Range>> ranges = new HashMap<>();
Map<ColumnHandle, Domain> predicates = tupleDomain.getDomains().get();
predicates
.entrySet()
.forEach(
entry -> {
ColumnHandle handle = entry.getKey();
if (handle instanceof HBaseColumnHandle) {
ranges.put(
((HBaseColumnHandle) handle).getOrdinal(),
entry.getValue().getValues().getRanges().getOrderedRanges());
}
});
predicates.entrySet().forEach(
entry -> {
ColumnHandle handle = entry.getKey();
if (handle instanceof HBaseColumnHandle) {
ranges.put(
((HBaseColumnHandle) handle).getOrdinal(),
entry.getValue().getValues().getRanges().getOrderedRanges());
}
});
List<HostAddress> hostAddresses = new ArrayList<>();
if (startEndKeys == null) {
throw new NullPointerException("null pointer found when getting splits for scan");
}
for (int i = 0; i < startEndKeys.getFirst().length; i++) {
String startRow = new String(startEndKeys.getFirst()[i]);
String endRow = new String(startEndKeys.getSecond()[i]);
// splitByChar read from hetu metastore, the default value is "0~9"
String splitByChar = hbaseConnection.getTable(tableHandle.getTableName()).getSplitByChar().get();
LOG.info("Create multi-splits by the first char of rowKey, table is " + hbaseTableName.getName() +
", the range of first char is : " + splitByChar);
List<StartAndEndKey> startAndEndRowKeys =
getStartAndEndKeys(splitByChar, Constants.START_END_KEYS_COUNT);
for (StartAndEndKey startAndEndRowKey : startAndEndRowKeys) {
splits.add(
new HBaseSplit(
tableHandle.getRowId(), tableHandle, hostAddresses, startRow, endRow, ranges, null, false));
tableHandle.getRowId(),
tableHandle,
hostAddresses,
String.valueOf(startAndEndRowKey.start),
startAndEndRowKey.end + Constants.ROWKEY_TAIL,
ranges,
false));
}
printSplits("Scan", splits);
return splits;
}
private List<HBaseSplit> getSplitsForBatchGet(TupleDomain<ColumnHandle> tupleDomain, HBaseTableHandle table)
/**
* In order to get more splits to improve concurrency of tableScan, we slice the split by different character.
* HBase server support to use startRow and EndRow to get scanner.
* for example, splitByChar is 0~2, we will generate multi-pairs
* the size of pairs will less than startEndKeysCount, so we will calculate the pair gap first.
* (startKey = 0, endKey = 0),(startKey = 1, endKey = 1),(startKey = 2, endKey = 2)
* splitByChar is 0~9, a~z
* (startKey = 0, endKey = 1),(startKey = 2, endKey = 3)……(startKey = y, endKey = z)
*
* @param splitByChar range of the rowKey, value is like 0~9,A~Z,a~z or a~z,0~9 ..
* @param startEndKeysCount max number of key pairs
* @return start and end rowKeys
*/
private List<StartAndEndKey> getStartAndEndKeys(String splitByChar, int startEndKeysCount)
{
List<StartAndEndKey> allRanges = Arrays.stream(splitByChar.split(","))
.map(StartAndEndKey::new).collect(Collectors.toList());
int rangeLength = 0;
for (StartAndEndKey range : allRanges) {
rangeLength += (Math.abs(range.end - range.start) + 1);
}
List<StartAndEndKey> startAndEndKeys = new ArrayList<>();
// rounding step value
int gap = (int) Math.rint((rangeLength + 0.0) / startEndKeysCount);
// generate start and end keys
allRanges.forEach(range -> {
int realGap = gap == 0 ? 1 : gap;
for (char index = range.start; index <= range.end; index += realGap) {
char end = (index + realGap > range.end) ? range.end : (char) (index + realGap - 1);
startAndEndKeys.add(new StartAndEndKey(index, end));
}
});
return startAndEndKeys;
}
/**
* If the predicate of sql includes "rowKey='xxx'" or "rowKey in ('xxx','xxx')",
* we can specify rowkey values in each split, then performance will be good.
*
* @param tupleDomain tupleDomain
* @param tableHandle tableHandle
* @return splits
*/
private List<HBaseSplit> getSplitsForBatchGet(TupleDomain<ColumnHandle> tupleDomain, HBaseTableHandle tableHandle)
{
List<HBaseSplit> splits = new ArrayList<>();
Domain rowIdDomain = null;
@ -150,13 +185,13 @@ public class HBaseSplitManager
ColumnHandle handle = entry.getKey();
if (handle instanceof HBaseColumnHandle) {
HBaseColumnHandle columnHandle = (HBaseColumnHandle) handle;
if (columnHandle.getOrdinal() == table.getRowIdOrdinal()) {
if (columnHandle.getOrdinal() == tableHandle.getRowIdOrdinal()) {
rowIdDomain = entry.getValue();
}
}
}
List<Range> rowIds = rowIdDomain != null ? rowIdDomain.getValues().getRanges().getOrderedRanges() : new ArrayList<>();
List<Range> rowIds = rowIdDomain != null ? rowIdDomain.getValues().getRanges().getOrderedRanges() : new ArrayList<>();
int maxSplitSize;
// Each split has at least 20 pieces of data, and the maximum number of splits is 30.
if (rowIds.size() / Constants.BATCHGET_SPLIT_RECORD_COUNT > Constants.BATCHGET_SPLIT_MAX_COUNT) {
@ -172,17 +207,48 @@ public class HBaseSplitManager
while (currentIndex < rangeSize) {
int endIndex = rangeSize - currentIndex > maxSplitSize ? (currentIndex + maxSplitSize) : rangeSize;
Map<Integer, List<Range>> splitRange = new HashMap<>();
splitRange.put(table.getRowIdOrdinal(), rowIds.subList(currentIndex, endIndex));
splits.add(new HBaseSplit(table.getRowId(), table, hostAddresses, null, null, splitRange, null, false));
splitRange.put(tableHandle.getRowIdOrdinal(), rowIds.subList(currentIndex, endIndex));
splits.add(new HBaseSplit(tableHandle.getRowId(), tableHandle, hostAddresses, null, null, splitRange, false));
currentIndex = endIndex;
}
for (HBaseSplit split : splits) {
if (LOG.isInfoEnabled()) {
LOG.info("Print Split: " + split.toString());
}
}
printSplits("Batch Get", splits);
return splits;
}
private void printSplits(String scanType, List<HBaseSplit> splits)
{
LOG.info("The final split count is " + splits.size() + ".");
for (HBaseSplit split : splits) {
LOG.info(scanType + ", Print Split: " + split.toString());
}
}
class StartAndEndKey
{
private final char start;
private final char end;
StartAndEndKey(char start, char end)
{
this.start = start;
this.end = end;
}
StartAndEndKey(String startAndEnd)
{
String[] pairs = startAndEnd.split(Constants.SEPARATOR_START_END_KEY);
this.start = pairs[0].charAt(0);
this.end = pairs[1].charAt(0);
}
@Override
public String toString()
{
return "StartAndEndKey{" +
"start=" + start +
", end=" + end +
'}';
}
}
}

73
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/utils/Constants.java
View File

@ -61,21 +61,26 @@ public class Constants
*/
public static final String SEPARATOR = ":";
/**
* SEPARATOR_START_END_KEY
*/
public static final String SEPARATOR_START_END_KEY = "~";
/**
* START_END_ROW_KEYS_COUNT
*/
public static final int START_END_KEYS_COUNT = 100;
/**
* ROWKEY_TAIL
*/
public static final String ROWKEY_TAIL = "|";
/**
* DEFAULT
*/
public static final String DEFAULT = "default";
/**
* ARRAY
*/
public static final String ARRAY = "ARRYA [ ";
/**
* apostrophe
*/
public static final String APOSTROPHE = "'";
/**
* 2
*/
@ -86,21 +91,6 @@ public class Constants
*/
public static final int NUMBER3 = 3;
/**
* 4
*/
public static final int NUMBER4 = 4;
/**
* 1024
*/
public static final int NUMBER1024 = 1024;
/**
* -1
*/
public static final int NUMBER_NEGATIVE_1 = -1;
/**
* "no such transaction: %s"
*/
@ -114,22 +104,12 @@ public class Constants
/**
* SCAN_CACHING_SIZE
*/
public static final int SCAN_CACHING_SIZE = 5000;
public static final int SCAN_CACHING_SIZE = 10000;
/**
* constant string
* PUT_BATCH_SIZE
*/
public static final String S_SCHEMA = "schema";
/**
* constant string
*/
public static final String S_TABLE = "table";
/**
* constant string
*/
public static final String S_COLUMNS = "columns";
public static final int PUT_BATCH_SIZE = 10000;
/**
* constant string
@ -159,7 +139,7 @@ public class Constants
/**
* constant string
*/
public static final String S_NAME = "name";
public static final String S_SPLIT_BY_CHAR = "splitByChar";
/**
* constant string
@ -171,21 +151,11 @@ public class Constants
*/
public static final String S_QUALIFIER = "qualifier";
/**
* constant string
*/
public static final String S_TYPE = "type";
/**
* constant string
*/
public static final String S_ORDINAL = "ordinal";
/**
* constant string
*/
public static final String S_COMMENT = "comment";
/**
* constant string
*/
@ -196,11 +166,6 @@ public class Constants
*/
public static final String S_TRUE = "true";
/**
* constant string
*/
public static final String S_FLASE = "false";
/**
* constant string
*/

5
hetu-hbase/src/main/java/io/hetu/core/plugin/hbase/utils/Utils.java
View File

@ -94,10 +94,7 @@ public class Utils
}
}
}
catch (ClassNotFoundException e) {
LOG.error("createTypeByName failed... cause by : %s", e);
}
catch (IllegalAccessException e) {
catch (ClassNotFoundException | IllegalAccessException e) {
LOG.error("createTypeByName failed... cause by : %s", e);
}

6
hetu-hbase/src/test/java/io/hetu/core/plugin/hbase/TestHBase.java
View File

@ -62,6 +62,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import static io.prestosql.spi.connector.ConnectorPageSink.NOT_BLOCKED;
import static java.lang.String.format;
@ -379,7 +380,7 @@ public class TestHBase
Map<Integer, List<Range>> ranges = new HashMap<>();
HBaseSplit split =
new HBaseSplit(
"rowkey", TestUtils.createHBaseTableHandle(), hostAddressList, null, null, ranges, null, true);
"rowkey", TestUtils.createHBaseTableHandle(), hostAddressList, null, null, ranges, false);
HBaseRecordSetProvider hrsp = new HBaseRecordSetProvider(hconn);
RecordSet rs =
@ -406,7 +407,8 @@ public class TestHBase
0,
hconn.getTable("hbase.test_table").getColumns(),
hconn.getTable("hbase.test_table").getSerializerClassName(),
Optional.of("test_table"));
Optional.of("test_table"),
OptionalLong.empty());
if (insertHandler instanceof ConnectorInsertTableHandle) {
ConnectorPageSink cps =
hpsp.createPageSink(

15
hetu-hbase/src/test/java/io/hetu/core/plugin/hbase/TestQuery.java
View File

@ -44,6 +44,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.UUID;
import static io.prestosql.spi.type.BigintType.BIGINT;
@ -90,8 +91,7 @@ public class TestQuery
"startrow",
"endrow",
new HashMap<>(),
null,
true);
false);
recordSet =
new HBaseRecordSet(
hconn, session, split, TestUtils.createHBaseTableHandle(), TestUtils.createColumnList());
@ -123,7 +123,8 @@ public class TestQuery
"",
TestUtils.createTupleDomain(1),
TestUtils.createColumnList(),
0);
0,
OptionalLong.empty());
HBaseSplit hBasesplit =
new HBaseSplit(
"rowKey",
@ -132,8 +133,7 @@ public class TestQuery
"startrow",
"endrow",
new HashMap<>(),
null,
true);
false);
HBaseRecordSet rSet = new HBaseRecordSet(hconn, session, hBasesplit, tableHandle, TestUtils.createColumnList());
rSet.cursor();
}
@ -157,7 +157,8 @@ public class TestQuery
"",
null,
list,
0);
0,
OptionalLong.empty());
// case ABOVE
Map<Integer, List<Range>> ranges = new HashMap<>();
@ -168,7 +169,7 @@ public class TestQuery
ranges.put(0, range);
HBaseSplit hBasesplit =
new HBaseSplit(
"rowkey", tableHandle, new ArrayList<HostAddress>(1), "1", "12345678", ranges, null, true);
"rowkey", tableHandle, new ArrayList<HostAddress>(1), "1", "12345678", ranges, false);
HBaseRecordSet rSet = new HBaseRecordSet(hconn, session, hBasesplit, tableHandle, list);
rSet.getFiltersFromDomains(ranges);

8
hetu-hbase/src/test/java/io/hetu/core/plugin/hbase/client/TestUtils.java
View File

@ -35,6 +35,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import static io.hetu.core.plugin.hbase.utils.ValueSetUtils.createValueSet;
import static io.prestosql.spi.type.VarcharType.VARCHAR;
@ -139,7 +140,8 @@ public class TestUtils
false,
"io.hetu.core.plugin.hbase.utils.serializers.StringRowSerializer",
Optional.of("test_table"),
"");
"",
OptionalLong.empty());
}
/**
@ -154,7 +156,8 @@ public class TestUtils
false,
"StringRowSerializer",
Optional.of(table),
"");
"",
OptionalLong.empty());
}
/**
@ -256,6 +259,7 @@ public class TestUtils
external,
Optional.of(serializerClassName),
Optional.of(table),
Optional.of(""),
Optional.of(""));
hTableMetaMemory.put(key, hBaseTable);
}

8
hetu-hbase/src/test/java/io/hetu/core/plugin/hbase/connector/TestHBaseConnector.java
View File

@ -427,7 +427,8 @@ public class TestHBaseConnector
false,
Optional.of("StringRowSerializer"),
Optional.of(""),
Optional.of("table"));
Optional.of("table"),
Optional.of(""));
// table is null
try {
hconn.dropTable(hBaseTable);
@ -492,7 +493,8 @@ public class TestHBaseConnector
false,
Optional.of("StringRowSerializer"),
Optional.of(""),
Optional.of("test_table"));
Optional.of("test_table"),
Optional.of(""));
try {
method.invoke(hconn, hBaseTable, "rowkey");
}
@ -581,7 +583,7 @@ public class TestHBaseConnector
hConnector.getPageSourceProvider();
hConnector.getPageSinkProvider();
hConnector.getPageSinkProvider();
assertEquals(hConnector.getTableProperties().size(), 7);
assertEquals(hConnector.getTableProperties().size(), 8);
assertEquals(hConnector.getColumnProperties().size(), 2);
}
}

7
hetu-hbase/src/test/java/io/hetu/core/plugin/hbase/connector/TestHBaseTableHandle.java
View File

@ -17,6 +17,7 @@ package io.hetu.core.plugin.hbase.connector;
import org.testng.annotations.Test;
import java.util.Optional;
import java.util.OptionalLong;
import static org.testng.Assert.assertEquals;
@ -47,7 +48,8 @@ public class TestHBaseTableHandle
false,
serializerClassName,
Optional.of(table),
"");
"",
OptionalLong.empty());
assertEquals(
"HBaseTableHandle{schema=hbase, table=table-1, rowId=rowkey, internal=false, "
+ "serializerClassName=io.hetu.core.plugin.hbase."
@ -88,7 +90,8 @@ public class TestHBaseTableHandle
false,
serializerClassName,
Optional.of(table),
"");
"",
OptionalLong.empty());
assertEquals(true, hBaseTableHandle.equals(hBaseTableHandle2));
}

1
hetu-hbase/src/test/java/io/hetu/core/plugin/hbase/metadata/TestHBaseTable.java
View File

@ -47,6 +47,7 @@ public class TestHBaseTable
false,
Optional.of("StringRowSerializer"),
Optional.of("table"),
Optional.of(""),
Optional.of(""));
}

9
hetu-hbase/src/test/java/io/hetu/core/plugin/hbase/metadata/TestHetuHBaseMetastore.java
View File

@ -66,7 +66,8 @@ public class TestHetuHBaseMetastore
false,
serializerClassName,
Optional.of(""),
Optional.of("testTable"));
Optional.of("testTable"),
Optional.of(""));
metaStore.addHBaseTable(testHBaseTable1);
@ -79,7 +80,8 @@ public class TestHetuHBaseMetastore
false,
serializerClassName,
Optional.of(""),
Optional.of("testTable2"));
Optional.of("testTable2"),
Optional.of(""));
metaStore.addHBaseTable(testHBaseTable2);
// test get all tables
@ -100,7 +102,8 @@ public class TestHetuHBaseMetastore
false,
serializerClassName,
Optional.of(""),
Optional.of("testTable3"));
Optional.of("testTable3"),
Optional.of(""));
metaStore.renameHBaseTable(testHBaseTable3, "hbase.testTable1");
hBaseTables = metaStore.getAllHBaseTables();
assertFalse(hBaseTables.containsKey("hbase.testTable1"));

3
hetu-hbase/src/test/java/io/hetu/core/plugin/hbase/metadata/TestingHetuMetastore.java
View File

@ -61,7 +61,8 @@ public class TestingHetuMetastore
false,
Optional.of("io.hetu.core.plugin.hbase.utils.serializers.StringRowSerializer"),
Optional.empty(),
Optional.of("hbase:test_table"));
Optional.of("hbase:test_table"),
Optional.empty());
private final TestingMySqlServer mySqlServer;
private final HetuHBaseMetastore metaStore;

4
hetu-hbase/src/test/java/io/hetu/core/plugin/hbase/split/TestHBaseSplit.java
View File

@ -44,18 +44,14 @@ public class TestHBaseSplit
"startrow",
"endrow",
new HashMap<>(),
null,
true);
assertEquals(0, split.getRanges().size());
assertEquals(TestUtils.createHBaseTableHandle(), split.getTableHandle());
String className = "io.hetu.core.plugin.hbase.split.HBaseSplit";
assertEquals(className, split.getInfo().toString().substring(0, className.length()));
assertEquals(true, split.isRemotelyAccessible());
assertEquals("rowKey", split.getRowKeyName());
assertEquals("endrow", split.getEndRow());
assertEquals(true, split.isRandomSplit());
assertEquals(null, split.getRegionInfo());
assertEquals(0, split.getAddresses().size());
}
}

11
hetu-hbase/src/test/java/io/hetu/core/plugin/hbase/split/TestHbaseSplitManager.java
View File

@ -24,7 +24,9 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
import java.util.NoSuchElementException;
import java.util.Optional;
import java.util.OptionalLong;
import static org.testng.Assert.assertEquals;
@ -70,10 +72,10 @@ public class TestHbaseSplitManager
{
try {
hsm.getSplits(null, null, TestUtils.createHBaseTableHandle(), null);
throw new NullPointerException("testSplitManager : failed");
throw new NoSuchElementException("No value present");
}
catch (NullPointerException e) {
assertEquals(e.toString(), "java.lang.NullPointerException: null pointer found when getting splits for scan");
catch (NoSuchElementException e) {
assertEquals(e.toString(), "java.util.NoSuchElementException: No value present");
}
}
@ -94,7 +96,8 @@ public class TestHbaseSplitManager
"",
TestUtils.createTupleDomain(1),
TestUtils.createColumnList(),
0);
0,
OptionalLong.empty());
hsm.getSplits(null, null, tableHandle, null);
}