You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by li...@apache.org on 2017/05/23 11:36:57 UTC
[04/16] kylin git commit: KYLIN-2632 Refactor REST API for better
convention
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server-base/src/main/java/org/apache/kylin/rest/service/QueryServiceV2.java
----------------------------------------------------------------------
diff --git a/server-base/src/main/java/org/apache/kylin/rest/service/QueryServiceV2.java b/server-base/src/main/java/org/apache/kylin/rest/service/QueryServiceV2.java
new file mode 100644
index 0000000..3699998
--- /dev/null
+++ b/server-base/src/main/java/org/apache/kylin/rest/service/QueryServiceV2.java
@@ -0,0 +1,516 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+package org.apache.kylin.rest.service;
+
+import static org.apache.kylin.common.util.CheckUtil.checkCondition;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+import javax.sql.DataSource;
+
+import org.apache.calcite.avatica.ColumnMetaData;
+import org.apache.commons.lang.StringUtils;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.QueryContext;
+import org.apache.kylin.common.debug.BackdoorToggles;
+import org.apache.kylin.common.exceptions.ResourceLimitExceededException;
+import org.apache.kylin.common.util.SetThreadName;
+import org.apache.kylin.cube.CubeManager;
+import org.apache.kylin.metadata.model.DataModelDesc;
+import org.apache.kylin.metadata.model.JoinDesc;
+import org.apache.kylin.metadata.model.JoinTableDesc;
+import org.apache.kylin.metadata.model.ModelDimensionDesc;
+import org.apache.kylin.metadata.model.TableRef;
+import org.apache.kylin.metadata.project.ProjectInstance;
+import org.apache.kylin.query.relnode.OLAPContext;
+import org.apache.kylin.query.util.QueryUtil;
+import org.apache.kylin.rest.constant.Constant;
+import org.apache.kylin.rest.exception.BadRequestException;
+import org.apache.kylin.rest.exception.InternalErrorException;
+import org.apache.kylin.rest.metrics.QueryMetricsFacade;
+import org.apache.kylin.rest.model.ColumnMetaWithType;
+import org.apache.kylin.rest.model.SelectedColumnMeta;
+import org.apache.kylin.rest.model.TableMetaWithType;
+import org.apache.kylin.rest.msg.Message;
+import org.apache.kylin.rest.msg.MsgPicker;
+import org.apache.kylin.rest.request.PrepareSqlRequest;
+import org.apache.kylin.rest.request.SQLRequest;
+import org.apache.kylin.rest.response.SQLResponse;
+import org.apache.kylin.rest.util.TableauInterceptor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.security.core.GrantedAuthority;
+import org.springframework.security.core.context.SecurityContextHolder;
+import org.springframework.stereotype.Component;
+
+import com.google.common.collect.Lists;
+
+import net.sf.ehcache.Cache;
+import net.sf.ehcache.Element;
+
+/**
+ * Created by luwei on 17-4-24.
+ */
+@Component("queryServiceV2")
+public class QueryServiceV2 extends QueryService {
+ private static final Logger logger = LoggerFactory.getLogger(QueryServiceV2.class);
+
+ @Autowired
+ @Qualifier("cacheService")
+ private CacheService cacheService;
+
+ @Autowired
+ @Qualifier("modelMgmtServiceV2")
+ private ModelServiceV2 modelServiceV2;
+
+ public SQLResponse doQueryWithCache(SQLRequest sqlRequest) {
+ Message msg = MsgPicker.getMsg();
+
+ KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
+ String serverMode = kylinConfig.getServerMode();
+ if (!(Constant.SERVER_MODE_QUERY.equals(serverMode.toLowerCase()) || Constant.SERVER_MODE_ALL.equals(serverMode.toLowerCase()))) {
+ throw new BadRequestException(String.format(msg.getQUERY_NOT_ALLOWED(), serverMode));
+ }
+ if (StringUtils.isBlank(sqlRequest.getProject())) {
+ throw new BadRequestException(msg.getEMPTY_PROJECT_NAME());
+ }
+
+ if (sqlRequest.getBackdoorToggles() != null)
+ BackdoorToggles.addToggles(sqlRequest.getBackdoorToggles());
+
+ final QueryContext queryContext = QueryContext.current();
+
+ try (SetThreadName ignored = new SetThreadName("Query %s", queryContext.getQueryId())) {
+ String sql = sqlRequest.getSql();
+ String project = sqlRequest.getProject();
+ logger.info("Using project: " + project);
+ logger.info("The original query: " + sql);
+
+ if (!sql.toLowerCase().contains("select")) {
+ logger.debug("Directly return exception as not supported");
+ throw new BadRequestException(msg.getNOT_SUPPORTED_SQL());
+ }
+
+ long startTime = System.currentTimeMillis();
+
+ SQLResponse sqlResponse = null;
+ boolean queryCacheEnabled = checkCondition(kylinConfig.isQueryCacheEnabled(), "query cache disabled in KylinConfig") && //
+ checkCondition(!BackdoorToggles.getDisableCache(), "query cache disabled in BackdoorToggles");
+
+ if (queryCacheEnabled) {
+ sqlResponse = searchQueryInCache(sqlRequest);
+ }
+
+ try {
+ if (null == sqlResponse) {
+ sqlResponse = query(sqlRequest);
+
+ long durationThreshold = kylinConfig.getQueryDurationCacheThreshold();
+ long scanCountThreshold = kylinConfig.getQueryScanCountCacheThreshold();
+ long scanBytesThreshold = kylinConfig.getQueryScanBytesCacheThreshold();
+ sqlResponse.setDuration(System.currentTimeMillis() - startTime);
+ logger.info("Stats of SQL response: isException: {}, duration: {}, total scan count {}", //
+ String.valueOf(sqlResponse.getIsException()), String.valueOf(sqlResponse.getDuration()), String.valueOf(sqlResponse.getTotalScanCount()));
+ if (checkCondition(queryCacheEnabled, "query cache is disabled") //
+ && checkCondition(!sqlResponse.getIsException(), "query has exception") //
+ && checkCondition(sqlResponse.getDuration() > durationThreshold || sqlResponse.getTotalScanCount() > scanCountThreshold || sqlResponse.getTotalScanBytes() > scanBytesThreshold, //
+ "query is too lightweight with duration: {} (threshold {}), scan count: {} (threshold {}), scan bytes: {} (threshold {})", sqlResponse.getDuration(), durationThreshold, sqlResponse.getTotalScanCount(), scanCountThreshold, sqlResponse.getTotalScanBytes(), scanBytesThreshold)
+ && checkCondition(sqlResponse.getResults().size() < kylinConfig.getLargeQueryThreshold(), "query response is too large: {} ({})", sqlResponse.getResults().size(), kylinConfig.getLargeQueryThreshold())) {
+ cacheManager.getCache(SUCCESS_QUERY_CACHE).put(new Element(sqlRequest, sqlResponse));
+ }
+
+ } else {
+ sqlResponse.setDuration(System.currentTimeMillis() - startTime);
+ sqlResponse.setTotalScanCount(0);
+ sqlResponse.setTotalScanBytes(0);
+ }
+
+ checkQueryAuth(sqlResponse);
+
+ } catch (Throwable e) { // calcite may throw AssertError
+ logger.error("Exception when execute sql", e);
+ String errMsg = QueryUtil.makeErrorMsgUserFriendly(e);
+
+ sqlResponse = new SQLResponse(null, null, 0, true, errMsg);
+ sqlResponse.setTotalScanCount(queryContext.getScannedRows());
+ sqlResponse.setTotalScanBytes(queryContext.getScannedBytes());
+
+ if (queryCacheEnabled && e.getCause() != null && e.getCause() instanceof ResourceLimitExceededException) {
+ Cache exceptionCache = cacheManager.getCache(EXCEPTION_QUERY_CACHE);
+ exceptionCache.put(new Element(sqlRequest, sqlResponse));
+ }
+ }
+
+ logQuery(sqlRequest, sqlResponse);
+
+ QueryMetricsFacade.updateMetrics(sqlRequest, sqlResponse);
+
+ if (sqlResponse.getIsException())
+ throw new InternalErrorException(sqlResponse.getExceptionMessage());
+
+ return sqlResponse;
+
+ } finally {
+ BackdoorToggles.cleanToggles();
+ QueryContext.reset();
+ }
+ }
+
+ public SQLResponse query(SQLRequest sqlRequest) throws Exception {
+ try {
+ final String user = SecurityContextHolder.getContext().getAuthentication().getName();
+ badQueryDetector.queryStart(Thread.currentThread(), sqlRequest, user);
+
+ return queryWithSqlMassage(sqlRequest);
+
+ } finally {
+ badQueryDetector.queryEnd(Thread.currentThread());
+ }
+ }
+
+ private SQLResponse queryWithSqlMassage(SQLRequest sqlRequest) throws Exception {
+ String userInfo = SecurityContextHolder.getContext().getAuthentication().getName();
+ final Collection<? extends GrantedAuthority> grantedAuthorities = SecurityContextHolder.getContext().getAuthentication().getAuthorities();
+ for (GrantedAuthority grantedAuthority : grantedAuthorities) {
+ userInfo += ",";
+ userInfo += grantedAuthority.getAuthority();
+ }
+
+ SQLResponse fakeResponse = TableauInterceptor.tableauIntercept(sqlRequest.getSql());
+ if (null != fakeResponse) {
+ logger.debug("Return fake response, is exception? " + fakeResponse.getIsException());
+ return fakeResponse;
+ }
+
+ String correctedSql = QueryUtil.massageSql(sqlRequest.getSql(), sqlRequest.getLimit(), sqlRequest.getOffset());
+ if (!correctedSql.equals(sqlRequest.getSql())) {
+ logger.info("The corrected query: " + correctedSql);
+
+ //CAUTION: should not change sqlRequest content!
+ //sqlRequest.setSql(correctedSql);
+ }
+
+ // add extra parameters into olap context, like acceptPartial
+ Map<String, String> parameters = new HashMap<String, String>();
+ parameters.put(OLAPContext.PRM_USER_AUTHEN_INFO, userInfo);
+ parameters.put(OLAPContext.PRM_ACCEPT_PARTIAL_RESULT, String.valueOf(sqlRequest.isAcceptPartial()));
+ OLAPContext.setParameters(parameters);
+ // force clear the query context before a new query
+ OLAPContext.clearThreadLocalContexts();
+
+ return execute(correctedSql, sqlRequest);
+
+ }
+
+ /**
+ * @param correctedSql
+ * @param sqlRequest
+ * @return
+ * @throws Exception
+ */
+ private SQLResponse execute(String correctedSql, SQLRequest sqlRequest) throws Exception {
+ Connection conn = null;
+ Statement stat = null;
+ ResultSet resultSet = null;
+
+ List<List<String>> results = Lists.newArrayList();
+ List<SelectedColumnMeta> columnMetas = Lists.newArrayList();
+
+ try {
+ conn = cacheService.getOLAPDataSource(sqlRequest.getProject()).getConnection();
+
+ if (sqlRequest instanceof PrepareSqlRequest) {
+ PreparedStatement preparedState = conn.prepareStatement(correctedSql);
+ processStatementAttr(preparedState, sqlRequest);
+
+ for (int i = 0; i < ((PrepareSqlRequest) sqlRequest).getParams().length; i++) {
+ setParam(preparedState, i + 1, ((PrepareSqlRequest) sqlRequest).getParams()[i]);
+ }
+
+ resultSet = preparedState.executeQuery();
+ } else {
+ stat = conn.createStatement();
+ processStatementAttr(stat, sqlRequest);
+ resultSet = stat.executeQuery(correctedSql);
+ }
+
+ ResultSetMetaData metaData = resultSet.getMetaData();
+ int columnCount = metaData.getColumnCount();
+
+ // Fill in selected column meta
+ for (int i = 1; i <= columnCount; ++i) {
+ columnMetas.add(new SelectedColumnMeta(metaData.isAutoIncrement(i), metaData.isCaseSensitive(i), metaData.isSearchable(i), metaData.isCurrency(i), metaData.isNullable(i), metaData.isSigned(i), metaData.getColumnDisplaySize(i), metaData.getColumnLabel(i), metaData.getColumnName(i), metaData.getSchemaName(i), metaData.getCatalogName(i), metaData.getTableName(i), metaData.getPrecision(i), metaData.getScale(i), metaData.getColumnType(i), metaData.getColumnTypeName(i), metaData.isReadOnly(i), metaData.isWritable(i), metaData.isDefinitelyWritable(i)));
+ }
+
+ // fill in results
+ while (resultSet.next()) {
+ List<String> oneRow = Lists.newArrayListWithCapacity(columnCount);
+ for (int i = 0; i < columnCount; i++) {
+ oneRow.add((resultSet.getString(i + 1)));
+ }
+
+ results.add(oneRow);
+ }
+ } finally {
+ close(resultSet, stat, conn);
+ }
+
+ boolean isPartialResult = false;
+ String cube = "";
+ StringBuilder sb = new StringBuilder("Processed rows for each storageContext: ");
+ if (OLAPContext.getThreadLocalContexts() != null) { // contexts can be null in case of 'explain plan for'
+ for (OLAPContext ctx : OLAPContext.getThreadLocalContexts()) {
+ if (ctx.realization != null) {
+ isPartialResult |= ctx.storageContext.isPartialResultReturned();
+ cube = ctx.realization.getName();
+ sb.append(ctx.storageContext.getProcessedRowCount()).append(" ");
+ }
+ }
+ }
+ logger.info(sb.toString());
+
+ SQLResponse response = new SQLResponse(columnMetas, results, cube, 0, false, null, isPartialResult);
+ response.setTotalScanCount(QueryContext.current().getScannedRows());
+ response.setTotalScanBytes(QueryContext.current().getScannedBytes());
+
+ return response;
+ }
+
+ /**
+ * @param preparedState
+ * @param param
+ * @throws SQLException
+ */
+ private void setParam(PreparedStatement preparedState, int index, PrepareSqlRequest.StateParam param) throws SQLException {
+ boolean isNull = (null == param.getValue());
+
+ Class<?> clazz;
+ try {
+ clazz = Class.forName(param.getClassName());
+ } catch (ClassNotFoundException e) {
+ throw new InternalErrorException(e);
+ }
+
+ ColumnMetaData.Rep rep = ColumnMetaData.Rep.of(clazz);
+
+ switch (rep) {
+ case PRIMITIVE_CHAR:
+ case CHARACTER:
+ case STRING:
+ preparedState.setString(index, isNull ? null : String.valueOf(param.getValue()));
+ break;
+ case PRIMITIVE_INT:
+ case INTEGER:
+ preparedState.setInt(index, isNull ? 0 : Integer.valueOf(param.getValue()));
+ break;
+ case PRIMITIVE_SHORT:
+ case SHORT:
+ preparedState.setShort(index, isNull ? 0 : Short.valueOf(param.getValue()));
+ break;
+ case PRIMITIVE_LONG:
+ case LONG:
+ preparedState.setLong(index, isNull ? 0 : Long.valueOf(param.getValue()));
+ break;
+ case PRIMITIVE_FLOAT:
+ case FLOAT:
+ preparedState.setFloat(index, isNull ? 0 : Float.valueOf(param.getValue()));
+ break;
+ case PRIMITIVE_DOUBLE:
+ case DOUBLE:
+ preparedState.setDouble(index, isNull ? 0 : Double.valueOf(param.getValue()));
+ break;
+ case PRIMITIVE_BOOLEAN:
+ case BOOLEAN:
+ preparedState.setBoolean(index, !isNull && Boolean.parseBoolean(param.getValue()));
+ break;
+ case PRIMITIVE_BYTE:
+ case BYTE:
+ preparedState.setByte(index, isNull ? 0 : Byte.valueOf(param.getValue()));
+ break;
+ case JAVA_UTIL_DATE:
+ case JAVA_SQL_DATE:
+ preparedState.setDate(index, isNull ? null : java.sql.Date.valueOf(param.getValue()));
+ break;
+ case JAVA_SQL_TIME:
+ preparedState.setTime(index, isNull ? null : Time.valueOf(param.getValue()));
+ break;
+ case JAVA_SQL_TIMESTAMP:
+ preparedState.setTimestamp(index, isNull ? null : Timestamp.valueOf(param.getValue()));
+ break;
+ default:
+ preparedState.setObject(index, isNull ? null : param.getValue());
+ }
+ }
+
+ public List<TableMetaWithType> getMetadataV2(String project) throws SQLException, IOException {
+ return getMetadataV2(getCubeManager(), project, true);
+ }
+
+ protected List<TableMetaWithType> getMetadataV2(CubeManager cubeMgr, String project, boolean cubedOnly) throws SQLException, IOException {
+ //Message msg = MsgPicker.getMsg();
+
+ Connection conn = null;
+ ResultSet columnMeta = null;
+ List<TableMetaWithType> tableMetas = null;
+ Map<String, TableMetaWithType> tableMap = null;
+ Map<String, ColumnMetaWithType> columnMap = null;
+ if (StringUtils.isBlank(project)) {
+ return Collections.emptyList();
+ }
+ ResultSet JDBCTableMeta = null;
+ try {
+ DataSource dataSource = cacheService.getOLAPDataSource(project);
+ conn = dataSource.getConnection();
+ DatabaseMetaData metaData = conn.getMetaData();
+
+ JDBCTableMeta = metaData.getTables(null, null, null, null);
+
+ tableMetas = new LinkedList<TableMetaWithType>();
+ tableMap = new HashMap<String, TableMetaWithType>();
+ columnMap = new HashMap<String, ColumnMetaWithType>();
+ while (JDBCTableMeta.next()) {
+ String catalogName = JDBCTableMeta.getString(1);
+ String schemaName = JDBCTableMeta.getString(2);
+
+ // Not every JDBC data provider offers full 10 columns, e.g., PostgreSQL has only 5
+ TableMetaWithType tblMeta = new TableMetaWithType(catalogName == null ? Constant.FakeCatalogName : catalogName, schemaName == null ? Constant.FakeSchemaName : schemaName, JDBCTableMeta.getString(3), JDBCTableMeta.getString(4), JDBCTableMeta.getString(5), null, null, null, null, null);
+
+ if (!cubedOnly || getProjectManager().isExposedTable(project, schemaName + "." + tblMeta.getTABLE_NAME())) {
+ tableMetas.add(tblMeta);
+ tableMap.put(tblMeta.getTABLE_SCHEM() + "#" + tblMeta.getTABLE_NAME(), tblMeta);
+ }
+ }
+
+ columnMeta = metaData.getColumns(null, null, null, null);
+
+ while (columnMeta.next()) {
+ String catalogName = columnMeta.getString(1);
+ String schemaName = columnMeta.getString(2);
+
+ // kylin(optiq) is not strictly following JDBC specification
+ ColumnMetaWithType colmnMeta = new ColumnMetaWithType(catalogName == null ? Constant.FakeCatalogName : catalogName, schemaName == null ? Constant.FakeSchemaName : schemaName, columnMeta.getString(3), columnMeta.getString(4), columnMeta.getInt(5), columnMeta.getString(6), columnMeta.getInt(7), getInt(columnMeta.getString(8)), columnMeta.getInt(9), columnMeta.getInt(10), columnMeta.getInt(11), columnMeta.getString(12), columnMeta.getString(13), getInt(columnMeta.getString(14)), getInt(columnMeta.getString(15)), columnMeta.getInt(16), columnMeta.getInt(17), columnMeta.getString(18), columnMeta.getString(19), columnMeta.getString(20), columnMeta.getString(21), getShort(columnMeta.getString(22)), columnMeta.getString(23));
+
+ if (!cubedOnly || getProjectManager().isExposedColumn(project, schemaName + "." + colmnMeta.getTABLE_NAME(), colmnMeta.getCOLUMN_NAME())) {
+ tableMap.get(colmnMeta.getTABLE_SCHEM() + "#" + colmnMeta.getTABLE_NAME()).addColumn(colmnMeta);
+ columnMap.put(colmnMeta.getTABLE_SCHEM() + "#" + colmnMeta.getTABLE_NAME() + "#" + colmnMeta.getCOLUMN_NAME(), colmnMeta);
+ }
+ }
+
+ } finally {
+ close(columnMeta, null, conn);
+ if (JDBCTableMeta != null) {
+ JDBCTableMeta.close();
+ }
+ }
+
+ ProjectInstance projectInstance = getProjectManager().getProject(project);
+ for (String modelName : projectInstance.getModels()) {
+ DataModelDesc dataModelDesc = modelServiceV2.listAllModels(modelName, project).get(0);
+ if (dataModelDesc.getStatus() == null) {
+
+ // update table type: FACT
+ for (TableRef factTable : dataModelDesc.getFactTables()) {
+ String factTableName = factTable.getTableIdentity().replace('.', '#');
+ if (tableMap.containsKey(factTableName)) {
+ tableMap.get(factTableName).getTYPE().add(TableMetaWithType.tableTypeEnum.FACT);
+ } else {
+ // should be used after JDBC exposes all tables and columns
+ // throw new BadRequestException(msg.getTABLE_META_INCONSISTENT());
+ }
+ }
+
+ // update table type: LOOKUP
+ for (TableRef lookupTable : dataModelDesc.getLookupTables()) {
+ String lookupTableName = lookupTable.getTableIdentity().replace('.', '#');
+ if (tableMap.containsKey(lookupTableName)) {
+ tableMap.get(lookupTableName).getTYPE().add(TableMetaWithType.tableTypeEnum.LOOKUP);
+ } else {
+ // throw new BadRequestException(msg.getTABLE_META_INCONSISTENT());
+ }
+ }
+
+ // update column type: PK and FK
+ for (JoinTableDesc joinTableDesc : dataModelDesc.getJoinTables()) {
+ JoinDesc joinDesc = joinTableDesc.getJoin();
+ for (String pk : joinDesc.getPrimaryKey()) {
+ String columnIdentity = (dataModelDesc.findTable(pk.substring(0, pk.indexOf("."))).getTableIdentity() + pk.substring(pk.indexOf("."))).replace('.', '#');
+ if (columnMap.containsKey(columnIdentity)) {
+ columnMap.get(columnIdentity).getTYPE().add(ColumnMetaWithType.columnTypeEnum.PK);
+ } else {
+ // throw new BadRequestException(msg.getCOLUMN_META_INCONSISTENT());
+ }
+ }
+
+ for (String fk : joinDesc.getForeignKey()) {
+ String columnIdentity = (dataModelDesc.findTable(fk.substring(0, fk.indexOf("."))).getTableIdentity() + fk.substring(fk.indexOf("."))).replace('.', '#');
+ if (columnMap.containsKey(columnIdentity)) {
+ columnMap.get(columnIdentity).getTYPE().add(ColumnMetaWithType.columnTypeEnum.FK);
+ } else {
+ // throw new BadRequestException(msg.getCOLUMN_META_INCONSISTENT());
+ }
+ }
+ }
+
+ // update column type: DIMENSION AND MEASURE
+ List<ModelDimensionDesc> dimensions = dataModelDesc.getDimensions();
+ for (ModelDimensionDesc dimension : dimensions) {
+ for (String column : dimension.getColumns()) {
+ String columnIdentity = (dataModelDesc.findTable(dimension.getTable()).getTableIdentity() + "." + column).replace('.', '#');
+ if (columnMap.containsKey(columnIdentity)) {
+ columnMap.get(columnIdentity).getTYPE().add(ColumnMetaWithType.columnTypeEnum.DIMENSION);
+ } else {
+ // throw new BadRequestException(msg.getCOLUMN_META_INCONSISTENT());
+ }
+
+ }
+ }
+
+ String[] measures = dataModelDesc.getMetrics();
+ for (String measure : measures) {
+ String columnIdentity = (dataModelDesc.findTable(measure.substring(0, measure.indexOf("."))).getTableIdentity() + measure.substring(measure.indexOf("."))).replace('.', '#');
+ if (columnMap.containsKey(columnIdentity)) {
+ columnMap.get(columnIdentity).getTYPE().add(ColumnMetaWithType.columnTypeEnum.MEASURE);
+ } else {
+ // throw new BadRequestException(msg.getCOLUMN_META_INCONSISTENT());
+ }
+ }
+ }
+ }
+
+ return tableMetas;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server-base/src/main/java/org/apache/kylin/rest/service/StreamingService.java
----------------------------------------------------------------------
diff --git a/server-base/src/main/java/org/apache/kylin/rest/service/StreamingService.java b/server-base/src/main/java/org/apache/kylin/rest/service/StreamingService.java
index 6f473e2..adae67c 100644
--- a/server-base/src/main/java/org/apache/kylin/rest/service/StreamingService.java
+++ b/server-base/src/main/java/org/apache/kylin/rest/service/StreamingService.java
@@ -25,7 +25,9 @@ import java.util.List;
import org.apache.commons.lang3.StringUtils;
import org.apache.kylin.metadata.streaming.StreamingConfig;
import org.apache.kylin.rest.constant.Constant;
-import org.apache.kylin.rest.exception.InternalErrorException;
+import org.apache.kylin.rest.exception.BadRequestException;
+import org.apache.kylin.rest.msg.Message;
+import org.apache.kylin.rest.msg.MsgPicker;
import org.springframework.security.access.prepost.PostFilter;
import org.springframework.stereotype.Component;
@@ -64,8 +66,10 @@ public class StreamingService extends BasicService {
}
public StreamingConfig createStreamingConfig(StreamingConfig config) throws IOException {
+ Message msg = MsgPicker.getMsg();
+
if (getStreamingManager().getStreamingConfig(config.getName()) != null) {
- throw new InternalErrorException("The streamingConfig named " + config.getName() + " already exists");
+ throw new BadRequestException(String.format(msg.getSTREAMING_CONFIG_ALREADY_EXIST(), config.getName()));
}
StreamingConfig streamingConfig = getStreamingManager().saveStreamingConfig(config);
return streamingConfig;
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server-base/src/main/java/org/apache/kylin/rest/service/TableService.java
----------------------------------------------------------------------
diff --git a/server-base/src/main/java/org/apache/kylin/rest/service/TableService.java b/server-base/src/main/java/org/apache/kylin/rest/service/TableService.java
index d4cb854..9f9b541 100644
--- a/server-base/src/main/java/org/apache/kylin/rest/service/TableService.java
+++ b/server-base/src/main/java/org/apache/kylin/rest/service/TableService.java
@@ -53,6 +53,7 @@ import org.apache.kylin.source.kafka.config.KafkaConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.stereotype.Component;
@@ -62,15 +63,19 @@ public class TableService extends BasicService {
private static final Logger logger = LoggerFactory.getLogger(TableService.class);
@Autowired
+ @Qualifier("modelMgmtService")
private ModelService modelService;
@Autowired
+ @Qualifier("projectService")
private ProjectService projectService;
@Autowired
+ @Qualifier("streamingMgmtService")
private StreamingService streamingService;
@Autowired
+ @Qualifier("kafkaMgmtService")
private KafkaConfigService kafkaConfigService;
public List<TableDesc> getTableDescByProject(String project, boolean withExt) throws IOException {
@@ -100,7 +105,7 @@ public class TableService extends BasicService {
return result;
}
- private void unLoadHiveTable(String tableName) throws IOException {
+ protected void unLoadHiveTable(String tableName) throws IOException {
tableName = normalizeHiveTableName(tableName);
MetadataManager metaMgr = MetadataManager.getInstance(KylinConfig.getInstanceFromEnv());
metaMgr.removeSourceTable(tableName);
@@ -111,7 +116,7 @@ public class TableService extends BasicService {
getProjectManager().addTableDescToProject(tables, project);
}
- private void removeTableFromProject(String tableName, String projectName) throws IOException {
+ protected void removeTableFromProject(String tableName, String projectName) throws IOException {
tableName = normalizeHiveTableName(tableName);
getProjectManager().removeTableDescFromProject(tableName, projectName);
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server-base/src/main/java/org/apache/kylin/rest/service/TableServiceV2.java
----------------------------------------------------------------------
diff --git a/server-base/src/main/java/org/apache/kylin/rest/service/TableServiceV2.java b/server-base/src/main/java/org/apache/kylin/rest/service/TableServiceV2.java
new file mode 100644
index 0000000..5c3eeb3
--- /dev/null
+++ b/server-base/src/main/java/org/apache/kylin/rest/service/TableServiceV2.java
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+package org.apache.kylin.rest.service;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.kylin.engine.mr.common.HadoopShellExecutable;
+import org.apache.kylin.engine.mr.common.MapReduceExecutable;
+import org.apache.kylin.job.execution.DefaultChainedExecutable;
+import org.apache.kylin.job.execution.ExecutableManager;
+import org.apache.kylin.job.execution.ExecutableState;
+import org.apache.kylin.metadata.MetadataManager;
+import org.apache.kylin.metadata.model.TableDesc;
+import org.apache.kylin.metadata.model.TableExtDesc;
+import org.apache.kylin.metadata.streaming.StreamingConfig;
+import org.apache.kylin.rest.constant.Constant;
+import org.apache.kylin.rest.exception.BadRequestException;
+import org.apache.kylin.rest.msg.Message;
+import org.apache.kylin.rest.msg.MsgPicker;
+import org.apache.kylin.source.hive.cardinality.HiveColumnCardinalityJob;
+import org.apache.kylin.source.hive.cardinality.HiveColumnCardinalityUpdateJob;
+import org.apache.kylin.source.kafka.config.KafkaConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.security.access.prepost.PreAuthorize;
+import org.springframework.security.core.context.SecurityContextHolder;
+import org.springframework.stereotype.Component;
+
+import com.google.common.collect.Sets;
+
+/**
+ * Created by luwei on 17-4-24.
+ */
+@Component("tableServiceV2")
+public class TableServiceV2 extends TableService {
+
+ private static final Logger logger = LoggerFactory.getLogger(TableServiceV2.class);
+
+ @Autowired
+ @Qualifier("modelMgmtServiceV2")
+ private ModelServiceV2 modelServiceV2;
+
+ @Autowired
+ @Qualifier("projectServiceV2")
+ private ProjectServiceV2 projectServiceV2;
+
+ @Autowired
+ @Qualifier("streamingMgmtService")
+ private StreamingService streamingService;
+
+ @Autowired
+ @Qualifier("kafkaMgmtService")
+ private KafkaConfigService kafkaConfigService;
+
+ @PreAuthorize(Constant.ACCESS_HAS_ROLE_MODELER + " or " + Constant.ACCESS_HAS_ROLE_ADMIN)
+ public void calculateCardinalityIfNotPresent(String[] tables, String submitter) throws IOException {
+ MetadataManager metaMgr = getMetadataManager();
+ ExecutableManager exeMgt = ExecutableManager.getInstance(getConfig());
+ for (String table : tables) {
+ TableExtDesc tableExtDesc = metaMgr.getTableExt(table);
+ String jobID = tableExtDesc.getJodID();
+ if (null == jobID || ExecutableState.RUNNING != exeMgt.getOutput(jobID).getState()) {
+ calculateCardinality(table, submitter);
+ }
+ }
+ }
+
+ /**
+ * Generate cardinality for table This will trigger a hadoop job
+ * The result will be merged into table exd info
+ *
+ * @param tableName
+ */
+ @PreAuthorize(Constant.ACCESS_HAS_ROLE_MODELER + " or " + Constant.ACCESS_HAS_ROLE_ADMIN)
+ public void calculateCardinality(String tableName, String submitter) throws IOException {
+ Message msg = MsgPicker.getMsg();
+
+ tableName = normalizeHiveTableName(tableName);
+ TableDesc table = getMetadataManager().getTableDesc(tableName);
+ final TableExtDesc tableExt = getMetadataManager().getTableExt(tableName);
+ if (table == null) {
+ BadRequestException e = new BadRequestException(String.format(msg.getTABLE_DESC_NOT_FOUND(), tableName));
+ logger.error("Cannot find table descriptor " + tableName, e);
+ throw e;
+ }
+
+ DefaultChainedExecutable job = new DefaultChainedExecutable();
+ //make sure the job could be scheduled when the DistributedScheduler is enable.
+ job.setParam("segmentId", tableName);
+ job.setName("Hive Column Cardinality calculation for table '" + tableName + "'");
+ job.setSubmitter(submitter);
+
+ String outPath = getConfig().getHdfsWorkingDirectory() + "cardinality/" + job.getId() + "/" + tableName;
+ String param = "-table " + tableName + " -output " + outPath;
+
+ MapReduceExecutable step1 = new MapReduceExecutable();
+
+ step1.setMapReduceJobClass(HiveColumnCardinalityJob.class);
+ step1.setMapReduceParams(param);
+ step1.setParam("segmentId", tableName);
+
+ job.addTask(step1);
+
+ HadoopShellExecutable step2 = new HadoopShellExecutable();
+
+ step2.setJobClass(HiveColumnCardinalityUpdateJob.class);
+ step2.setJobParams(param);
+ step2.setParam("segmentId", tableName);
+ job.addTask(step2);
+ tableExt.setJodID(job.getId());
+ getMetadataManager().saveTableExt(tableExt);
+
+ getExecutableManager().addJob(job);
+ }
+
+ /**
+ * table may referenced by several projects, and kylin only keep one copy of meta for each table,
+ * that's why we have two if statement here.
+ * @param tableName
+ * @param project
+ * @return
+ */
+ @PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN)
+ public boolean unLoadHiveTableV2(String tableName, String project) throws IOException {
+ Message msg = MsgPicker.getMsg();
+
+ boolean rtn = false;
+ int tableType = 0;
+
+ //remove streaming info
+ tableName = normalizeHiveTableName(tableName);
+ TableDesc desc = getMetadataManager().getTableDesc(tableName);
+ if (desc == null)
+ return false;
+ tableType = desc.getSourceType();
+
+ if (!modelServiceV2.isTableInModel(tableName, project)) {
+ removeTableFromProject(tableName, project);
+ rtn = true;
+ } else {
+ List<String> models = modelServiceV2.getModelsUsingTable(tableName, project);
+ throw new BadRequestException(String.format(msg.getTABLE_IN_USE_BY_MODEL(), models));
+ }
+
+ if (!projectServiceV2.isTableInAnyProject(tableName) && !modelServiceV2.isTableInAnyModel(tableName)) {
+ unLoadHiveTable(tableName);
+ rtn = true;
+ }
+
+ if (tableType == 1 && !projectServiceV2.isTableInAnyProject(tableName) && !modelServiceV2.isTableInAnyModel(tableName)) {
+ StreamingConfig config = null;
+ KafkaConfig kafkaConfig = null;
+ try {
+ config = streamingService.getStreamingManager().getStreamingConfig(tableName);
+ kafkaConfig = kafkaConfigService.getKafkaConfig(tableName);
+ streamingService.dropStreamingConfig(config);
+ kafkaConfigService.dropKafkaConfig(kafkaConfig);
+ rtn = true;
+ } catch (Exception e) {
+ rtn = false;
+ logger.error(e.getLocalizedMessage(), e);
+ }
+ }
+ return rtn;
+ }
+
+ public Map<String, String[]> loadHiveTables(String[] tableNames, String project, boolean isNeedProfile) throws IOException {
+ String submitter = SecurityContextHolder.getContext().getAuthentication().getName();
+ Map<String, String[]> result = new HashMap<String, String[]>();
+
+ String[] loaded = loadHiveTablesToProject(tableNames, project);
+ result.put("result.loaded", loaded);
+ Set<String> allTables = new HashSet<String>();
+ for (String tableName : tableNames) {
+ allTables.add(normalizeHiveTableName(tableName));
+ }
+ for (String loadedTableName : loaded) {
+ allTables.remove(loadedTableName);
+ }
+ String[] unloaded = new String[allTables.size()];
+ allTables.toArray(unloaded);
+ result.put("result.unloaded", unloaded);
+ if (isNeedProfile) {
+ calculateCardinalityIfNotPresent(loaded, submitter);
+ }
+ return result;
+ }
+
+ public Map<String, String[]> unloadHiveTables(String[] tableNames, String project) throws IOException {
+ Set<String> unLoadSuccess = Sets.newHashSet();
+ Set<String> unLoadFail = Sets.newHashSet();
+ Map<String, String[]> result = new HashMap<String, String[]>();
+
+ for (String tableName : tableNames) {
+ if (unLoadHiveTableV2(tableName, project)) {
+ unLoadSuccess.add(tableName);
+ } else {
+ unLoadFail.add(tableName);
+ }
+ }
+
+ result.put("result.unload.success", (String[]) unLoadSuccess.toArray(new String[unLoadSuccess.size()]));
+ result.put("result.unload.fail", (String[]) unLoadFail.toArray(new String[unLoadFail.size()]));
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server-base/src/main/java/org/apache/kylin/rest/service/UserService.java
----------------------------------------------------------------------
diff --git a/server-base/src/main/java/org/apache/kylin/rest/service/UserService.java b/server-base/src/main/java/org/apache/kylin/rest/service/UserService.java
index 2d8e006..f37d447 100644
--- a/server-base/src/main/java/org/apache/kylin/rest/service/UserService.java
+++ b/server-base/src/main/java/org/apache/kylin/rest/service/UserService.java
@@ -31,6 +31,9 @@ import org.apache.kylin.common.persistence.ResourceStore;
import org.apache.kylin.common.persistence.Serializer;
import org.apache.kylin.common.util.JsonUtil;
import org.apache.kylin.rest.constant.Constant;
+import org.apache.kylin.rest.exception.InternalErrorException;
+import org.apache.kylin.rest.msg.Message;
+import org.apache.kylin.rest.msg.MsgPicker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.access.prepost.PreAuthorize;
@@ -55,7 +58,6 @@ public class UserService implements UserDetailsManager {
@PostConstruct
public void init() throws IOException {
aclStore = ResourceStore.getStore(KylinConfig.getInstanceFromEnv());
- logger.debug("UserService init");
}
@Override
@@ -73,7 +75,7 @@ public class UserService implements UserDetailsManager {
aclStore.putResource(id, new UserInfo(user), 0, UserInfoSerializer.getInstance());
logger.debug("update user : {}", user.getUsername());
} catch (IOException e) {
- throw new RuntimeException(e);
+ throw new InternalErrorException(e);
}
}
@@ -84,7 +86,7 @@ public class UserService implements UserDetailsManager {
aclStore.deleteResource(id);
logger.debug("delete user : {}", userName);
} catch (IOException e) {
- throw new RuntimeException(e);
+ throw new InternalErrorException(e);
}
}
@@ -99,25 +101,26 @@ public class UserService implements UserDetailsManager {
logger.debug("judge user exist: {}", userName);
return aclStore.exists(getId(userName));
} catch (IOException e) {
- throw new RuntimeException(e);
+ throw new InternalErrorException(e);
}
}
@Override
public UserDetails loadUserByUsername(String userName) throws UsernameNotFoundException {
+ Message msg = MsgPicker.getMsg();
try {
UserInfo userInfo = aclStore.getResource(getId(userName), UserInfo.class, UserInfoSerializer.getInstance());
if (userInfo == null) {
- throw new UsernameNotFoundException("User:" + userName + " Not found");
+ throw new UsernameNotFoundException(String.format(msg.getUSER_NOT_FOUND(), userName));
}
logger.debug("load user : {}", userName);
return wrap(userInfo);
} catch (IOException e) {
- throw new RuntimeException(e);
+ throw new InternalErrorException(e);
}
}
- public List<String> listUserAuthorities() {
+ public List<String> listUserAuthorities() throws IOException {
List<String> all = new ArrayList<String>();
for (UserDetails user : listUsers()) {
for (GrantedAuthority auth : user.getAuthorities()) {
@@ -129,15 +132,11 @@ public class UserService implements UserDetailsManager {
return all;
}
- public List<UserDetails> listUsers() {
+ public List<UserDetails> listUsers() throws IOException {
List<UserDetails> all = new ArrayList<UserDetails>();
- try {
- List<UserInfo> userInfos = aclStore.getAllResources(DIR_PREFIX, UserInfo.class, UserInfoSerializer.getInstance());
- for (UserInfo info : userInfos) {
- all.add(wrap(info));
- }
- } catch (IOException e) {
- throw new RuntimeException("Failed to list users", e);
+ List<UserInfo> userInfos = aclStore.getAllResources(DIR_PREFIX, UserInfo.class, UserInfoSerializer.getInstance());
+ for (UserInfo info : userInfos) {
+ all.add(wrap(info));
}
return all;
}
@@ -146,7 +145,7 @@ public class UserService implements UserDetailsManager {
return DIR_PREFIX + userName;
}
- private User wrap(UserInfo userInfo) {
+ protected User wrap(UserInfo userInfo) {
if (userInfo == null)
return null;
List<GrantedAuthority> authorities = new ArrayList<>();
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server-base/src/main/java/org/apache/kylin/rest/util/ControllerSplitter.java
----------------------------------------------------------------------
diff --git a/server-base/src/main/java/org/apache/kylin/rest/util/ControllerSplitter.java b/server-base/src/main/java/org/apache/kylin/rest/util/ControllerSplitter.java
new file mode 100644
index 0000000..e043327
--- /dev/null
+++ b/server-base/src/main/java/org/apache/kylin/rest/util/ControllerSplitter.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kylin.rest.util;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.io.IOUtils;
+
+public class ControllerSplitter {
+
+ static File v1dir = new File("src/main/java/org/apache/kylin/rest/controller");
+ static File v2dir = new File("src/main/java/org/apache/kylin/rest/controller2");
+ static boolean dryRun = false;
+
+ public static void main(String[] args) throws IOException {
+
+ for (File f : v1dir.listFiles()) {
+ chopOff(f, "application/vnd.apache.kylin-v2+json");
+ }
+
+ for (File f : v2dir.listFiles()) {
+ chopOff(f, "application/json");
+ }
+ }
+
+ private static void chopOff(File f, String annoPtn) throws IOException {
+
+ System.out.println("Processing " + f);
+
+ FileInputStream is = new FileInputStream(f);
+ List<String> lines = IOUtils.readLines(is, "UTF-8");
+ is.close();
+ List<String> outLines = new ArrayList<>(lines.size());
+
+ boolean del = false;
+ for (String l : lines) {
+ if (l.startsWith(" @") && l.contains(annoPtn))
+ del = true;
+
+ if (del)
+ System.out.println("x " + l);
+ else
+ outLines.add(l);
+
+ if (del && l.startsWith(" }"))
+ del = false;
+ }
+
+ if (!dryRun && outLines.size() < lines.size()) {
+ FileOutputStream os = new FileOutputStream(f);
+ IOUtils.writeLines(outLines, "\n", os, "UTF-8");
+ os.close();
+ System.out.println("UPDATED " + f);
+ } else {
+ System.out.println("skipped");
+ }
+
+ System.out.println("============================================================================");
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/main/resources/applicationContext.xml
----------------------------------------------------------------------
diff --git a/server/src/main/resources/applicationContext.xml b/server/src/main/resources/applicationContext.xml
index 081dc53..100b202 100644
--- a/server/src/main/resources/applicationContext.xml
+++ b/server/src/main/resources/applicationContext.xml
@@ -49,7 +49,7 @@
<!-- Rest service binding -->
<bean
- class="org.springframework.web.servlet.mvc.annotation.DefaultAnnotationHandlerMapping"/>
+ class="org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerMapping"/>
<bean id="mappingJacksonHttpMessageConverter"
class="org.springframework.http.converter.json.MappingJackson2HttpMessageConverter"/>
@@ -59,7 +59,7 @@
class="org.springframework.http.converter.FormHttpMessageConverter"/>
<bean
- class="org.springframework.web.servlet.mvc.annotation.AnnotationMethodHandlerAdapter">
+ class="org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter">
<property name="messageConverters">
<list>
<ref bean="mappingJacksonHttpMessageConverter"/>
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/controller/AccessControllerTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/controller/AccessControllerTest.java b/server/src/test/java/org/apache/kylin/rest/controller/AccessControllerTest.java
index fea98cd..18fbd06 100644
--- a/server/src/test/java/org/apache/kylin/rest/controller/AccessControllerTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/controller/AccessControllerTest.java
@@ -32,6 +32,7 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.security.access.AccessDeniedException;
import org.springframework.security.authentication.TestingAuthenticationToken;
import org.springframework.security.core.Authentication;
@@ -62,12 +63,15 @@ public class AccessControllerTest extends ServiceTestBase implements AclEntityTy
private String ADMIN = "ADMIN";
@Autowired
+ @Qualifier("projectService")
ProjectService projectService;
@Autowired
+ @Qualifier("cubeMgmtService")
CubeService cubeService;
@Autowired
+ @Qualifier("accessService")
AccessService accessService;
@Before
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/controller/AdminControllerTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/controller/AdminControllerTest.java b/server/src/test/java/org/apache/kylin/rest/controller/AdminControllerTest.java
index e6b6dc0..823e68a 100644
--- a/server/src/test/java/org/apache/kylin/rest/controller/AdminControllerTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/controller/AdminControllerTest.java
@@ -27,6 +27,7 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
/**
* @author xduo
@@ -36,8 +37,11 @@ public class AdminControllerTest extends ServiceTestBase {
private AdminController adminController;
@Autowired
+ @Qualifier("adminService")
private AdminService adminService;
+
@Autowired
+ @Qualifier("cubeMgmtService")
private CubeService cubeService;
@Before
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/controller/CacheControllerTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/controller/CacheControllerTest.java b/server/src/test/java/org/apache/kylin/rest/controller/CacheControllerTest.java
index cda0d61..c2e21cc 100644
--- a/server/src/test/java/org/apache/kylin/rest/controller/CacheControllerTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/controller/CacheControllerTest.java
@@ -25,6 +25,7 @@ import org.apache.kylin.rest.service.ServiceTestBase;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
/**
* @author shaoshi
@@ -34,6 +35,7 @@ public class CacheControllerTest extends ServiceTestBase {
private CacheController cacheController;
@Autowired
+ @Qualifier("cacheService")
private CacheService cacheService;
@Before
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/controller/CubeControllerTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/controller/CubeControllerTest.java b/server/src/test/java/org/apache/kylin/rest/controller/CubeControllerTest.java
index 5c010ef..e67c238 100644
--- a/server/src/test/java/org/apache/kylin/rest/controller/CubeControllerTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/controller/CubeControllerTest.java
@@ -38,6 +38,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
+import org.springframework.beans.factory.annotation.Qualifier;
/**
* @author xduo
@@ -48,10 +49,15 @@ public class CubeControllerTest extends ServiceTestBase {
private CubeDescController cubeDescController;
@Autowired
+ @Qualifier("cubeMgmtService")
CubeService cubeService;
+
@Autowired
+ @Qualifier("jobService")
JobService jobService;
+
@Autowired
+ @Qualifier("streamingMgmtService")
StreamingService streamingService;
@Before
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/controller/JobControllerTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/controller/JobControllerTest.java b/server/src/test/java/org/apache/kylin/rest/controller/JobControllerTest.java
index 3cd994b..e87cecb 100644
--- a/server/src/test/java/org/apache/kylin/rest/controller/JobControllerTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/controller/JobControllerTest.java
@@ -41,6 +41,7 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
/**
* @author xduo
@@ -50,9 +51,11 @@ public class JobControllerTest extends ServiceTestBase {
private JobController jobSchedulerController;
private CubeController cubeController;
@Autowired
+ @Qualifier("jobService")
JobService jobService;
@Autowired
+ @Qualifier("cubeMgmtService")
CubeService cubeService;
private static final String CUBE_NAME = "new_job_controller";
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/controller/ProjectControllerTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/controller/ProjectControllerTest.java b/server/src/test/java/org/apache/kylin/rest/controller/ProjectControllerTest.java
index cd9a524..f805095 100644
--- a/server/src/test/java/org/apache/kylin/rest/controller/ProjectControllerTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/controller/ProjectControllerTest.java
@@ -33,6 +33,7 @@ import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.beans.factory.annotation.Qualifier;
/**
*/
@@ -41,6 +42,7 @@ public class ProjectControllerTest extends ServiceTestBase {
private ProjectController projectController;
@Autowired
+ @Qualifier("projectService")
ProjectService projectService;
@Before
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/controller/QueryControllerTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/controller/QueryControllerTest.java b/server/src/test/java/org/apache/kylin/rest/controller/QueryControllerTest.java
index 2043fef..d9eb3fa 100644
--- a/server/src/test/java/org/apache/kylin/rest/controller/QueryControllerTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/controller/QueryControllerTest.java
@@ -31,6 +31,7 @@ import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import net.sf.ehcache.CacheManager;
+import org.springframework.beans.factory.annotation.Qualifier;
/**
* @author xduo
@@ -39,7 +40,9 @@ public class QueryControllerTest extends ServiceTestBase {
private QueryController queryController;
@Autowired
+ @Qualifier("queryService")
QueryService queryService;
+
@Autowired
private CacheManager cacheManager;
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/service/AccessServiceTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/service/AccessServiceTest.java b/server/src/test/java/org/apache/kylin/rest/service/AccessServiceTest.java
index f27a121..481b0bf 100644
--- a/server/src/test/java/org/apache/kylin/rest/service/AccessServiceTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/service/AccessServiceTest.java
@@ -26,6 +26,7 @@ import org.apache.kylin.rest.security.AclPermissionFactory;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.security.acls.domain.PrincipalSid;
import org.springframework.security.acls.model.AccessControlEntry;
import org.springframework.security.acls.model.Acl;
@@ -37,6 +38,7 @@ import org.springframework.security.acls.model.Sid;
public class AccessServiceTest extends ServiceTestBase {
@Autowired
+ @Qualifier("accessService")
AccessService accessService;
@Test
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/service/CubeServiceTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/service/CubeServiceTest.java b/server/src/test/java/org/apache/kylin/rest/service/CubeServiceTest.java
index 59e96d6..a190d6d 100644
--- a/server/src/test/java/org/apache/kylin/rest/service/CubeServiceTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/service/CubeServiceTest.java
@@ -29,6 +29,7 @@ import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import com.fasterxml.jackson.core.JsonProcessingException;
+import org.springframework.beans.factory.annotation.Qualifier;
/**
* @author xduo
@@ -36,9 +37,11 @@ import com.fasterxml.jackson.core.JsonProcessingException;
public class CubeServiceTest extends ServiceTestBase {
@Autowired
+ @Qualifier("cubeMgmtService")
CubeService cubeService;
@Autowired
+ @Qualifier("cacheService")
private CacheService cacheService;
@Test
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/service/JobServiceTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/service/JobServiceTest.java b/server/src/test/java/org/apache/kylin/rest/service/JobServiceTest.java
index 4150808..0932748 100644
--- a/server/src/test/java/org/apache/kylin/rest/service/JobServiceTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/service/JobServiceTest.java
@@ -26,6 +26,7 @@ import org.apache.kylin.metadata.project.ProjectInstance;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
/**
* @author xduo
@@ -33,9 +34,11 @@ import org.springframework.beans.factory.annotation.Autowired;
public class JobServiceTest extends ServiceTestBase {
@Autowired
+ @Qualifier("jobService")
JobService jobService;
@Autowired
+ @Qualifier("cacheService")
private CacheService cacheService;
@Test
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/service/ModelServiceTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/service/ModelServiceTest.java b/server/src/test/java/org/apache/kylin/rest/service/ModelServiceTest.java
index 550aeb8..7e749a9 100644
--- a/server/src/test/java/org/apache/kylin/rest/service/ModelServiceTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/service/ModelServiceTest.java
@@ -35,10 +35,12 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
public class ModelServiceTest extends ServiceTestBase {
@Autowired
+ @Qualifier("modelMgmtService")
ModelService modelService;
@Rule
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/service/QueryServiceTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/service/QueryServiceTest.java b/server/src/test/java/org/apache/kylin/rest/service/QueryServiceTest.java
index ca4e34c..1e5a7a5 100644
--- a/server/src/test/java/org/apache/kylin/rest/service/QueryServiceTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/service/QueryServiceTest.java
@@ -28,6 +28,7 @@ import org.apache.kylin.rest.response.SQLResponse;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
/**
* @author xduo
@@ -35,9 +36,11 @@ import org.springframework.beans.factory.annotation.Autowired;
public class QueryServiceTest extends ServiceTestBase {
@Autowired
+ @Qualifier("queryService")
QueryService queryService;
@Autowired
+ @Qualifier("cacheService")
private CacheService cacheService;
@Test
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/service/ServiceTestBase.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/service/ServiceTestBase.java b/server/src/test/java/org/apache/kylin/rest/service/ServiceTestBase.java
index 7370c48..b45b27b 100644
--- a/server/src/test/java/org/apache/kylin/rest/service/ServiceTestBase.java
+++ b/server/src/test/java/org/apache/kylin/rest/service/ServiceTestBase.java
@@ -31,6 +31,7 @@ import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.security.authentication.TestingAuthenticationToken;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
@@ -48,6 +49,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
public class ServiceTestBase extends LocalFileMetadataTestCase {
@Autowired
+ @Qualifier("userService")
UserService userService;
@BeforeClass
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/server/src/test/java/org/apache/kylin/rest/service/UserServiceTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/service/UserServiceTest.java b/server/src/test/java/org/apache/kylin/rest/service/UserServiceTest.java
index 3b3a738..0d4b580 100644
--- a/server/src/test/java/org/apache/kylin/rest/service/UserServiceTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/service/UserServiceTest.java
@@ -18,6 +18,7 @@
package org.apache.kylin.rest.service;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@@ -25,6 +26,7 @@ import org.apache.kylin.rest.constant.Constant;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.security.core.GrantedAuthority;
import org.springframework.security.core.authority.SimpleGrantedAuthority;
import org.springframework.security.core.userdetails.User;
@@ -36,10 +38,11 @@ import org.springframework.security.core.userdetails.UserDetails;
public class UserServiceTest extends ServiceTestBase {
@Autowired
+ @Qualifier("userService")
UserService userService;
@Test
- public void testBasics() {
+ public void testBasics() throws IOException {
userService.deleteUser("ADMIN");
Assert.assertTrue(!userService.userExists("ADMIN"));
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
----------------------------------------------------------------------
diff --git a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
index e388074..bdbe321 100644
--- a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
+++ b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
@@ -61,8 +61,8 @@ import com.google.common.collect.Sets;
public class HiveMRInput implements IMRInput {
- public static String getTableNameForHCat(TableDesc table, boolean isFullTable) {
- String tableName = (table.isView() || isFullTable == false) ? table.getMaterializedName() : table.getName();
+ public static String getTableNameForHCat(TableDesc table) {
+ String tableName = (table.isView()) ? table.getMaterializedName() : table.getName();
return String.format("%s.%s", table.getDatabase(), tableName).toUpperCase();
}
@@ -73,12 +73,7 @@ public class HiveMRInput implements IMRInput {
@Override
public IMRTableInputFormat getTableInputFormat(TableDesc table) {
- return new HiveTableInputFormat(getTableNameForHCat(table, true));
- }
-
- @Override
- public IMRTableInputFormat getTableInputFormat(TableDesc table, boolean isFullTable) {
- return new HiveTableInputFormat(getTableNameForHCat(table, isFullTable));
+ return new HiveTableInputFormat(getTableNameForHCat(table));
}
@Override
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java
----------------------------------------------------------------------
diff --git a/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java b/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java
index 9033d67..ffd54db 100644
--- a/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java
+++ b/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java
@@ -63,7 +63,7 @@ public class ColumnCardinalityMapper<T> extends KylinMapper<T, Object, IntWritab
String tableName = conf.get(BatchConstants.CFG_TABLE_NAME);
tableDesc = MetadataManager.getInstance(config).getTableDesc(tableName);
- tableInputFormat = MRUtil.getTableInputFormat(tableDesc, true);
+ tableInputFormat = MRUtil.getTableInputFormat(tableDesc);
}
@Override
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/HiveColumnCardinalityJob.java
----------------------------------------------------------------------
diff --git a/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/HiveColumnCardinalityJob.java b/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/HiveColumnCardinalityJob.java
index a837681..f439ccb 100644
--- a/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/HiveColumnCardinalityJob.java
+++ b/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/HiveColumnCardinalityJob.java
@@ -87,7 +87,7 @@ public class HiveColumnCardinalityJob extends AbstractHadoopJob {
job.getConfiguration().set("mapreduce.output.fileoutputformat.compress", "false");
// Mapper
- IMRTableInputFormat tableInputFormat = MRUtil.getTableInputFormat(table, true);
+ IMRTableInputFormat tableInputFormat = MRUtil.getTableInputFormat(table);
tableInputFormat.configureJob(job);
job.setMapperClass(ColumnCardinalityMapper.class);
http://git-wip-us.apache.org/repos/asf/kylin/blob/73a78dbe/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
----------------------------------------------------------------------
diff --git a/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java b/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
index 4c140be..3323afb 100644
--- a/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
+++ b/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
@@ -76,11 +76,6 @@ public class KafkaMRInput implements IMRInput {
@Override
public IMRTableInputFormat getTableInputFormat(TableDesc table) {
- return getTableInputFormat(table, true);
- }
-
- @Override
- public IMRTableInputFormat getTableInputFormat(TableDesc table, boolean isFullTable) {
KafkaConfigManager kafkaConfigManager = KafkaConfigManager.getInstance(KylinConfig.getInstanceFromEnv());
KafkaConfig kafkaConfig = kafkaConfigManager.getKafkaConfig(table.getIdentity());
List<TblColRef> columns = Lists.transform(Arrays.asList(table.getColumns()), new Function<ColumnDesc, TblColRef>() {