database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[Feature-Variant](Variant Type) support variant type query and index (#27676)

Browse Source
This commit is contained in:
lihangyu
2023-11-29 10:37:28 +08:00
committed by GitHub
parent 3bc09e55f6
commit 7398c3daf1
433 changed files with 10443 additions and 879 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
fe/fe-core/src/main/java/org/apache/doris/analysis/Analyzer.java
View File

@ -91,6 +91,7 @@ import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.Iterator;
@ -124,6 +125,11 @@ public class Analyzer {
// map from lowercase qualified column name ("alias.col") to descriptor
private final Map<String, SlotDescriptor> slotRefMap = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
// Notice: it's case sensitive
// Variant column name -> Paths of sub columns
private final Map<String, Map<List<String>, SlotDescriptor>> subColumnSlotRefMap
= Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
// map from tuple id to list of conjuncts referencing tuple
private final Map<TupleId, List<ExprId>> tuplePredicates = Maps.newHashMap();
// map from slot id to list of conjuncts referencing slot
@ -922,7 +928,8 @@ public class Analyzer {
* @param colName
* @throws AnalysisException
*/
public SlotDescriptor registerColumnRef(TableName tblName, String colName) throws AnalysisException {
public SlotDescriptor registerColumnRef(TableName tblName, String colName, List<String> subColNames)
throws AnalysisException {
TupleDescriptor d;
TableName newTblName = tblName;
if (newTblName == null) {
@ -1004,11 +1011,55 @@ public class Analyzer {
newTblName == null ? d.getTable().getName() : newTblName.toString());
}
LOG.debug("register column ref table {}, colName {}, col {}", tblName, colName, col.toSql());
if (col.getType().isVariantType() || (subColNames != null && !subColNames.isEmpty())) {
if (!col.getType().isVariantType()) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_ILLEGAL_COLUMN_REFERENCE_ERROR,
Joiner.on(".").join(tblName.getTbl(), colName));
}
if (subColNames == null) {
// Root
subColNames = new ArrayList<String>();
}
String key = d.getAlias() + "." + col.getName();
if (subColumnSlotRefMap.get(key) == null) {
subColumnSlotRefMap.put(key, Maps.newTreeMap(
new Comparator<List<String>>() {
public int compare(List<String> lst1, List<String> lst2) {
Iterator<String> it1 = lst1.iterator();
Iterator<String> it2 = lst2.iterator();
while (it1.hasNext() && it2.hasNext()) {
int result = it1.next().compareTo(it2.next());
if (result != 0) {
return result;
}
}
return Integer.compare(lst1.size(), lst2.size());
}
}));
}
SlotDescriptor result = subColumnSlotRefMap.get(key).get(subColNames);
if (result != null) {
// avoid duplicate slots
return result;
}
result = globalState.descTbl.addSlotDescriptor(d);
LOG.debug("register slot descriptor {}", result);
result.setSubColLables(subColNames);
result.setColumn(col);
if (!subColNames.isEmpty()) {
result.setMaterializedColumnName(col.getName() + "." + String.join(".", subColNames));
}
result.setIsMaterialized(true);
result.setIsNullable(col.isAllowNull());
subColumnSlotRefMap.get(key).put(subColNames, result);
return result;
}
// Make column name case insensitive
String key = d.getAlias() + "." + col.getName();
SlotDescriptor result = slotRefMap.get(key);
if (result != null) {
result.setMultiRef(true);
return result;
}
result = globalState.descTbl.addSlotDescriptor(d);
@ -1032,7 +1083,6 @@ public class Analyzer {
String key = colName;
SlotDescriptor result = slotRefMap.get(key);
if (result != null) {
result.setMultiRef(true);
return result;
}
result = addSlotDescriptor(tupleDescriptor);
@ -1110,6 +1160,7 @@ public class Analyzer {
result.setStats(srcSlotDesc.getStats());
result.setType(srcSlotDesc.getType());
result.setIsNullable(srcSlotDesc.getIsNullable());
result.setSubColLables(srcSlotDesc.getSubColLables());
if (srcSlotDesc.getColumn() != null) {
result.setColumn(srcSlotDesc.getColumn());
}

4
fe/fe-core/src/main/java/org/apache/doris/analysis/Expr.java
View File

@ -1019,6 +1019,10 @@ public abstract class Expr extends TreeNode<Expr> implements ParseNode, Cloneabl
return toSql();
}
public List<String> toSubColumnLabel() {
return Lists.newArrayList();
}
// Convert this expr, including all children, to its Thrift representation.
public TExpr treeToThrift() {
TExpr result = new TExpr();

3
fe/fe-core/src/main/java/org/apache/doris/analysis/IndexDef.java
View File

@ -219,7 +219,8 @@ public class IndexDef {
colType = ((ArrayType) column.getType()).getItemType().getPrimitiveType();
}
if (!(colType.isDateType() || colType.isDecimalV2Type() || colType.isDecimalV3Type()
|| colType.isFixedPointType() || colType.isStringType() || colType == PrimitiveType.BOOLEAN)) {
|| colType.isFixedPointType() || colType.isStringType() || colType == PrimitiveType.BOOLEAN
|| colType.isVariantType())) {
throw new AnalysisException(colType + " is not supported in " + indexType.toString() + " index. "
+ "invalid column: " + indexColName);
} else if (indexType == IndexType.INVERTED

13
fe/fe-core/src/main/java/org/apache/doris/analysis/InlineViewRef.java
View File

@ -56,6 +56,7 @@ public class InlineViewRef extends TableRef {
// and column labels used in the query definition. Either all or none of the column
// labels must be overridden.
private List<String> explicitColLabels;
private List<List<String>> explicitSubColPath;
// ///////////////////////////////////////
// BEGIN: Members that need to be reset()
@ -97,6 +98,7 @@ public class InlineViewRef extends TableRef {
public InlineViewRef(String alias, QueryStmt queryStmt, List<String> colLabels) {
this(alias, queryStmt);
explicitColLabels = Lists.newArrayList(colLabels);
LOG.debug("inline view explicitColLabels {}", explicitColLabels);
}
/**
@ -153,6 +155,12 @@ public class InlineViewRef extends TableRef {
return queryStmt.getColLabels();
}
public List<List<String>> getSubColPath() {
if (explicitSubColPath != null) {
return explicitSubColPath;
}
return queryStmt.getSubColPath();
}
@Override
public void reset() {
@ -227,9 +235,12 @@ public class InlineViewRef extends TableRef {
// TODO: relax this a bit by allowing propagation out of the inline view (but
// not into it)
List<SlotDescriptor> slots = analyzer.changeSlotToNullableOfOuterJoinedTuples();
LOG.debug("inline view query {}", queryStmt.toSql());
for (int i = 0; i < getColLabels().size(); ++i) {
String colName = getColLabels().get(i);
SlotDescriptor slotDesc = analyzer.registerColumnRef(getAliasAsName(), colName);
LOG.debug("inline view register {}", colName);
SlotDescriptor slotDesc = analyzer.registerColumnRef(getAliasAsName(),
colName, getSubColPath().get(i));
Expr colExpr = queryStmt.getResultExprs().get(i);
if (queryStmt instanceof SelectStmt && ((SelectStmt) queryStmt).getValueList() != null) {
ValueList valueList = ((SelectStmt) queryStmt).getValueList();

2
fe/fe-core/src/main/java/org/apache/doris/analysis/InvertedIndexUtil.java
View File

@ -105,7 +105,7 @@ public class InvertedIndexUtil {
parser = INVERTED_INDEX_PARSER_NONE;
}
if (colType.isStringType()) {
if (colType.isStringType() || colType.isVariantType()) {
if (!(parser.equals(INVERTED_INDEX_PARSER_NONE)
|| parser.equals(INVERTED_INDEX_PARSER_STANDARD)
|| parser.equals(INVERTED_INDEX_PARSER_UNICODE)

2
fe/fe-core/src/main/java/org/apache/doris/analysis/QueryStmt.java
View File

@ -591,6 +591,8 @@ public abstract class QueryStmt extends StatementBase implements Queriable {
*/
public abstract ArrayList<String> getColLabels();
public abstract ArrayList<List<String>> getSubColPath();
/**
* Returns the materialized tuple ids of the output of this stmt.
* Used in case this stmt is part of an @InlineViewRef,

8
fe/fe-core/src/main/java/org/apache/doris/analysis/SelectListItem.java
View File

@ -22,6 +22,8 @@ package org.apache.doris.analysis;
import com.google.common.base.Preconditions;
import java.util.List;
public class SelectListItem {
private Expr expr;
// for "[name.]*"
@ -149,6 +151,12 @@ public class SelectListItem {
return "__" + expr.getExprName() + "_" + position;
}
public List<String> toSubColumnLabels() {
Preconditions.checkState(!isStar());
return expr.toSubColumnLabel();
}
public void setAlias(String alias) {
this.alias = alias;
}

18
fe/fe-core/src/main/java/org/apache/doris/analysis/SelectStmt.java
View File

@ -90,6 +90,7 @@ public class SelectStmt extends QueryStmt {
protected SelectList selectList;
private final ArrayList<String> colLabels; // lower case column labels
private final ArrayList<List<String>> subColPath; // case insensitive column labels
protected FromClause fromClause;
protected GroupByClause groupByClause;
private List<Expr> originalExpr;
@ -145,6 +146,7 @@ public class SelectStmt extends QueryStmt {
this.selectList = new SelectList();
this.fromClause = new FromClause();
this.colLabels = Lists.newArrayList();
this.subColPath = Lists.newArrayList();
}
public SelectStmt(
@ -171,6 +173,7 @@ public class SelectStmt extends QueryStmt {
this.havingClause = havingPredicate;
this.colLabels = Lists.newArrayList();
this.subColPath = Lists.newArrayList();
this.havingPred = null;
this.aggInfo = null;
this.sortInfo = null;
@ -191,6 +194,7 @@ public class SelectStmt extends QueryStmt {
other.havingClauseAfterAnalyzed != null ? other.havingClauseAfterAnalyzed.clone() : null;
colLabels = Lists.newArrayList(other.colLabels);
subColPath = Lists.newArrayList(other.subColPath);
aggInfo = (other.aggInfo != null) ? other.aggInfo.clone() : null;
analyticInfo = (other.analyticInfo != null) ? other.analyticInfo.clone() : null;
sqlString = (other.sqlString != null) ? other.sqlString : null;
@ -213,6 +217,7 @@ public class SelectStmt extends QueryStmt {
super.reset();
selectList.reset();
colLabels.clear();
subColPath.clear();
fromClause.reset();
if (whereClause != null) {
whereClause.reset();
@ -368,6 +373,12 @@ public class SelectStmt extends QueryStmt {
return colLabels;
}
@Override
public ArrayList<List<String>> getSubColPath() {
return subColPath;
}
public ExprSubstitutionMap getBaseTblSmap() {
return baseTblSmap;
}
@ -591,6 +602,7 @@ public class SelectStmt extends QueryStmt {
}
aliasSMap.put(aliasRef, item.getExpr().clone());
colLabels.add(columnLabel);
subColPath.add(item.toSubColumnLabels());
}
}
}
@ -632,6 +644,7 @@ public class SelectStmt extends QueryStmt {
resultExprs.add(rewriteQueryExprByMvColumnExpr(expr, analyzer));
}
colLabels.add("col_" + colLabels.size());
subColPath.add(expr.toSubColumnLabel());
}
}
// analyze valueList if exists
@ -816,7 +829,8 @@ public class SelectStmt extends QueryStmt {
LOG.debug("only support duplicate key or MOW model");
return false;
}
if (!olapTable.getEnableLightSchemaChange() || !Strings.isNullOrEmpty(olapTable.getStoragePolicy())) {
if (!olapTable.getEnableLightSchemaChange() || !Strings.isNullOrEmpty(olapTable.getStoragePolicy())
|| olapTable.hasVariantColumns()) {
return false;
}
if (getOrderByElements() != null) {
@ -1240,6 +1254,8 @@ public class SelectStmt extends QueryStmt {
slot.setTupleId(desc.getId());
resultExprs.add(rewriteQueryExprByMvColumnExpr(slot, analyzer));
colLabels.add(col.getName());
// empty sub lables
subColPath.add(Lists.newArrayList());
}
}

6
fe/fe-core/src/main/java/org/apache/doris/analysis/SetOperationStmt.java
View File

@ -797,6 +797,12 @@ public class SetOperationStmt extends QueryStmt {
return operands.get(0).getQueryStmt().getColLabels();
}
@Override
public ArrayList<List<String>> getSubColPath() {
Preconditions.checkState(operands.size() > 0);
return operands.get(0).getQueryStmt().getSubColPath();
}
@Override
public void setNeedToSql(boolean needToSql) {
super.setNeedToSql(needToSql);

39
fe/fe-core/src/main/java/org/apache/doris/analysis/SlotDescriptor.java
View File

@ -44,6 +44,12 @@ public class SlotDescriptor {
// for SlotRef.toSql() in the absence of a path
private String label;
// for variant column's sub column lables
private List<String> subColPath;
// materializedColumnName is the target name of a slot
// it could be either column name or a composed name for a variant
// subcolumn like `a.b.c`
private String materializedColumnName;
// Expr(s) materialized into this slot; multiple exprs for unions. Should be empty if
// path_ is set.
@ -64,7 +70,6 @@ public class SlotDescriptor {
private ColumnStats stats; // only set if 'column' isn't set
private boolean isAgg;
private boolean isMultiRef;
// If set to false, then such slots will be ignored during
// materialize them.Used to optimize to read less data and less memory usage
private boolean needMaterialize = true;
@ -77,7 +82,6 @@ public class SlotDescriptor {
this.isMaterialized = false;
this.isNullable = true;
this.isAgg = false;
this.isMultiRef = false;
}
public SlotDescriptor(SlotId id, TupleDescriptor parent, SlotDescriptor src) {
@ -95,14 +99,6 @@ public class SlotDescriptor {
this.sourceExprs.add(new SlotRef(src));
}
public boolean isMultiRef() {
return isMultiRef;
}
public void setMultiRef(boolean isMultiRef) {
this.isMultiRef = isMultiRef;
}
public boolean getIsAgg() {
return isAgg;
}
@ -123,6 +119,14 @@ public class SlotDescriptor {
return id;
}
public void setSubColLables(List<String> subColPath) {
this.subColPath = subColPath;
}
public List<String> getSubColLables() {
return this.subColPath;
}
public TupleDescriptor getParent() {
return parent;
}
@ -212,6 +216,10 @@ public class SlotDescriptor {
this.stats = stats;
}
public void setMaterializedColumnName(String name) {
this.materializedColumnName = name;
}
public ColumnStats getStats() {
if (stats == null) {
if (column != null) {
@ -295,9 +303,10 @@ public class SlotDescriptor {
public TSlotDescriptor toThrift() {
// Non-nullable slots will have 0 for the byte offset and -1 for the bit mask
String colName = materializedColumnName != null ? materializedColumnName :
((column != null) ? column.getNonShadowName() : "");
TSlotDescriptor tSlotDescriptor = new TSlotDescriptor(id.asInt(), parent.getId().asInt(), type.toThrift(), -1,
byteOffset, 0, getIsNullable() ? 0 : -1,
((column != null) ? column.getNonShadowName() : ""), slotIdx,
byteOffset, 0, getIsNullable() ? 0 : -1, colName, slotIdx,
isMaterialized);
tSlotDescriptor.setNeedMaterialize(needMaterialize);
tSlotDescriptor.setIsAutoIncrement(isAutoInc);
@ -308,6 +317,9 @@ public class SlotDescriptor {
tSlotDescriptor.setIsKey(column.isKey());
tSlotDescriptor.setColDefaultValue(column.getDefaultValue());
}
if (subColPath != null) {
tSlotDescriptor.setColumnPaths(subColPath);
}
return tSlotDescriptor;
}
@ -318,7 +330,7 @@ public class SlotDescriptor {
return MoreObjects.toStringHelper(this).add("id", id.asInt()).add("parent", parentTupleId).add("col", colStr)
.add("type", typeStr).add("materialized", isMaterialized).add("byteSize", byteSize)
.add("byteOffset", byteOffset).add("slotIdx", slotIdx).add("nullable", getIsNullable())
.add("isAutoIncrement", isAutoInc).toString();
.add("isAutoIncrement", isAutoInc).add("subColPath", subColPath).toString();
}
@Override
@ -335,6 +347,7 @@ public class SlotDescriptor {
.append(", type=").append(type == null ? "null" : type.toSql())
.append(", nullable=").append(isNullable)
.append(", isAutoIncrement=").append(isAutoInc)
.append(", subColPath=").append(subColPath)
.append("}")
.toString();
}

34
fe/fe-core/src/main/java/org/apache/doris/analysis/SlotRef.java
View File

@ -57,6 +57,7 @@ public class SlotRef extends Expr {
private String col;
// Used in toSql
private String label;
private List<String> subColPath;
// results of analysis
protected SlotDescriptor desc;
@ -73,6 +74,14 @@ public class SlotRef extends Expr {
this.label = "`" + col + "`";
}
public SlotRef(TableName tblName, String col, List<String> subColPath) {
super();
this.tblName = tblName;
this.col = col;
this.label = "`" + col + "`";
this.subColPath = subColPath;
}
// C'tor for a "pre-analyzed" ref to slot that doesn't correspond to
// a table's column.
public SlotRef(SlotDescriptor desc) {
@ -86,6 +95,7 @@ public class SlotRef extends Expr {
if (this.type.equals(Type.CHAR)) {
this.type = Type.VARCHAR;
}
this.subColPath = desc.getSubColLables();
analysisDone();
}
@ -109,6 +119,7 @@ public class SlotRef extends Expr {
label = other.label;
desc = other.desc;
tupleId = other.tupleId;
subColPath = other.subColPath;
}
@Override
@ -202,7 +213,7 @@ public class SlotRef extends Expr {
@Override
public void analyzeImpl(Analyzer analyzer) throws AnalysisException {
desc = analyzer.registerColumnRef(tblName, col);
desc = analyzer.registerColumnRef(tblName, col, subColPath);
type = desc.getType();
if (this.type.equals(Type.CHAR)) {
this.type = Type.VARCHAR;
@ -229,6 +240,7 @@ public class SlotRef extends Expr {
helper.add("type", type.toSql());
helper.add("label", label);
helper.add("tblName", tblName != null ? tblName.toSql() : "null");
helper.add("subColPath", subColPath);
return helper.toString();
}
@ -315,6 +327,10 @@ public class SlotRef extends Expr {
return this.exprName.get();
}
public List<String> toSubColumnLabel() {
return subColPath;
}
@Override
protected void toThrift(TExprNode msg) {
msg.node_type = TExprNodeType.SLOT_REF;
@ -333,7 +349,14 @@ public class SlotRef extends Expr {
if (desc != null) {
return desc.getId().hashCode();
}
return Objects.hashCode((tblName == null ? "" : tblName.toSql() + "." + label).toLowerCase());
if (subColPath == null || subColPath.isEmpty()) {
return Objects.hashCode((tblName == null ? "" : tblName.toSql() + "." + label).toLowerCase());
}
int result = Objects.hashCode((tblName == null ? "" : tblName.toSql() + "." + label).toLowerCase());
for (String sublabel : subColPath) {
result = 31 * result + Objects.hashCode(sublabel);
}
return result;
}
@Override
@ -368,6 +391,13 @@ public class SlotRef extends Expr {
if (col != null && !col.equalsIgnoreCase(other.col)) {
return false;
}
if ((subColPath == null) != (other.subColPath == null)) {
return false;
}
if (subColPath != null
&& subColPath.equals(other.subColPath)) {
return false;
}
return true;
}

9
fe/fe-core/src/main/java/org/apache/doris/analysis/TupleDescriptor.java
View File

@ -177,6 +177,15 @@ public class TupleDescriptor {
return null;
}
public boolean hasVariantCol() {
for (SlotDescriptor slotDesc : slots) {
if (slotDesc.getColumn() != null && slotDesc.getColumn().getType().isVariantType()) {
return true;
}
}
return false;
}
public TableIf getTable() {
return table;
}

9
fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
View File

@ -2418,4 +2418,13 @@ public class OlapTable extends Table {
public boolean needReadLockWhenPlan() {
return true;
}
public boolean hasVariantColumns() {
for (Column column : getBaseSchema()) {
if (column.getType().isVariantType()) {
return true;
}
}
return false;
}
}

14
fe/fe-core/src/main/java/org/apache/doris/planner/MaterializedViewSelector.java
View File

@ -202,6 +202,17 @@ public class MaterializedViewSelector {
for (Map.Entry<Long, MaterializedIndexMeta> entry : candidateIndexIdToMeta.entrySet()) {
result.put(entry.getKey(), entry.getValue().getSchema());
}
// For query like `select v:a from tbl` when column v is variant type but v:a is not expicity
// in index, so the above check will filter all index. But we should at least choose the base
// index at present.TODO we should better handle it.
LOG.debug("result {}, has variant col {}, tuple {}", result,
analyzer.getTupleDesc(scanNode.getTupleId()).hasVariantCol(),
analyzer.getTupleDesc(scanNode.getTupleId()).toString());
if (result.keySet().size() == 0 && scanNode.getOlapTable()
.getBaseSchema().stream().anyMatch(column -> column.getType().isVariantType())) {
LOG.info("Using base schema");
result.put(scanNode.getOlapTable().getBaseIndexId(), scanNode.getOlapTable().getBaseSchema());
}
return result;
}
@ -577,7 +588,8 @@ public class MaterializedViewSelector {
candidateIndexSchema
.forEach(column -> indexColumnNames.add(CreateMaterializedViewStmt
.mvColumnBreaker(MaterializedIndexMeta.normalizeName(column.getName()))));
LOG.debug("candidateIndexSchema {}, indexColumnNames {}, queryColumnNames {}",
candidateIndexSchema, indexColumnNames, queryColumnNames);
// Rollup index have no define expr.
if (entry.getValue().getWhereClause() == null && indexExprs.isEmpty()
&& !indexColumnNames.containsAll(queryColumnNames)) {

161
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
View File

@ -17,9 +17,7 @@
package org.apache.doris.service;
import org.apache.doris.alter.SchemaChangeHandler;
import org.apache.doris.analysis.AbstractBackupTableRefClause;
import org.apache.doris.analysis.AddColumnsClause;
import org.apache.doris.analysis.AddPartitionClause;
import org.apache.doris.analysis.Analyzer;
import org.apache.doris.analysis.ColumnDef;
@ -40,7 +38,6 @@ import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.DatabaseIf;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.Index;
import org.apache.doris.catalog.MaterializedIndex;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
@ -106,8 +103,6 @@ import org.apache.doris.tablefunction.MetadataGenerator;
import org.apache.doris.task.StreamLoadTask;
import org.apache.doris.thrift.FrontendService;
import org.apache.doris.thrift.FrontendServiceVersion;
import org.apache.doris.thrift.TAddColumnsRequest;
import org.apache.doris.thrift.TAddColumnsResult;
import org.apache.doris.thrift.TAutoIncrementRangeRequest;
import org.apache.doris.thrift.TAutoIncrementRangeResult;
import org.apache.doris.thrift.TBackend;
@ -116,7 +111,6 @@ import org.apache.doris.thrift.TBeginTxnResult;
import org.apache.doris.thrift.TBinlog;
import org.apache.doris.thrift.TCheckAuthRequest;
import org.apache.doris.thrift.TCheckAuthResult;
import org.apache.doris.thrift.TColumn;
import org.apache.doris.thrift.TColumnDef;
import org.apache.doris.thrift.TColumnDesc;
import org.apache.doris.thrift.TCommitTxnRequest;
@ -236,15 +230,11 @@ import org.apache.logging.log4j.Logger;
import org.apache.thrift.TException;
import java.io.StringReader;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
@ -255,7 +245,6 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.IntSupplier;
import java.util.stream.Collectors;
// Frontend service used to serve all request for this frontend through
@ -467,156 +456,6 @@ public class FrontendServiceImpl implements FrontendService.Iface {
defaultVal, comment, true);
}
@Override
public TAddColumnsResult addColumns(TAddColumnsRequest request) throws TException {
String clientAddr = getClientAddrAsString();
LOG.debug("schema change clientAddr: {}, request: {}", clientAddr, request);
TStatus status = new TStatus(TStatusCode.OK);
List<TColumn> allColumns = new ArrayList<TColumn>();
Env env = Env.getCurrentEnv();
InternalCatalog catalog = env.getInternalCatalog();
int schemaVersion = 0;
try {
if (!env.isMaster()) {
status.setStatusCode(TStatusCode.ILLEGAL_STATE);
status.addToErrorMsgs("retry rpc request to master.");
TAddColumnsResult result = new TAddColumnsResult();
result.setStatus(status);
return result;
}
TableName tableName = new TableName("", request.getDbName(), request.getTableName());
if (request.getTableId() > 0) {
tableName = catalog.getTableNameByTableId(request.getTableId());
}
if (tableName == null) {
throw new MetaNotFoundException("table_id " + request.getTableId() + " does not exist");
}
Database db = catalog.getDbNullable(tableName.getDb());
if (db == null) {
throw new MetaNotFoundException("db " + tableName.getDb() + " does not exist");
}
List<TColumnDef> addColumns = request.getAddColumns();
boolean queryMode = false;
if (addColumns == null || addColumns.size() == 0) {
queryMode = true;
}
// rpc only olap table
OlapTable olapTable = (OlapTable) db.getTableOrMetaException(tableName.getTbl(), TableType.OLAP);
olapTable.writeLockOrMetaException();
try {
olapTable.checkNormalStateForAlter();
List<ColumnDef> columnDefs = new ArrayList<ColumnDef>();
// prepare columnDefs
for (TColumnDef tColumnDef : addColumns) {
if (request.isAllowTypeConflict()) {
// ignore column with same name
boolean hasSameNameColumn = false;
for (Column column : olapTable.getBaseSchema()) {
if (column.getName().equalsIgnoreCase(tColumnDef.getColumnDesc().getColumnName())) {
hasSameNameColumn = true;
}
}
// ignore this column
if (hasSameNameColumn) {
continue;
}
}
String comment = tColumnDef.getComment();
if (comment == null || comment.length() == 0) {
Instant ins = Instant.ofEpochSecond(System.currentTimeMillis() / 1000);
ZonedDateTime zdt = ins.atZone(ZoneId.systemDefault());
comment = "auto change " + zdt.toString();
}
TColumnDesc tColumnDesc = tColumnDef.getColumnDesc();
ColumnDef columnDef = initColumnfromThrift(tColumnDesc, comment);
columnDefs.add(columnDef);
}
if (!queryMode && !columnDefs.isEmpty()) {
// create AddColumnsClause
AddColumnsClause addColumnsClause = new AddColumnsClause(columnDefs, null, null);
addColumnsClause.analyze(null);
// index id -> index schema
Map<Long, LinkedList<Column>> indexSchemaMap = new HashMap<>();
// index id -> index col_unique_id supplier
Map<Long, IntSupplier> colUniqueIdSupplierMap = new HashMap<>();
for (Map.Entry<Long, List<Column>> entry : olapTable.getIndexIdToSchema(true).entrySet()) {
indexSchemaMap.put(entry.getKey(), new LinkedList<>(entry.getValue()));
IntSupplier colUniqueIdSupplier = null;
if (olapTable.getEnableLightSchemaChange()) {
colUniqueIdSupplier = new IntSupplier() {
public int pendingMaxColUniqueId = olapTable
.getIndexMetaByIndexId(entry.getKey()).getMaxColUniqueId();
@Override
public int getAsInt() {
pendingMaxColUniqueId++;
return pendingMaxColUniqueId;
}
};
}
colUniqueIdSupplierMap.put(entry.getKey(), colUniqueIdSupplier);
}
// 4. call schame change function, only for dynamic table feature.
SchemaChangeHandler schemaChangeHandler = new SchemaChangeHandler();
boolean lightSchemaChange = schemaChangeHandler.processAddColumns(
addColumnsClause, olapTable, indexSchemaMap, true, colUniqueIdSupplierMap);
if (lightSchemaChange) {
// for schema change add column optimize, direct modify table meta.
List<Index> newIndexes = olapTable.getCopiedIndexes();
long jobId = Env.getCurrentEnv().getNextId();
Env.getCurrentEnv().getSchemaChangeHandler().modifyTableLightSchemaChange(
"",
db, olapTable, indexSchemaMap, newIndexes, null, false, jobId, false);
} else {
throw new MetaNotFoundException("table_id "
+ request.getTableId() + " cannot light schema change through rpc.");
}
}
// 5. build all columns
for (Column column : olapTable.getBaseSchema()) {
allColumns.add(column.toThrift());
}
schemaVersion = olapTable.getBaseSchemaVersion();
} catch (Exception e) {
LOG.warn("got exception add columns: ", e);
status.setStatusCode(TStatusCode.INTERNAL_ERROR);
status.addToErrorMsgs(e.getMessage());
} finally {
olapTable.writeUnlock();
}
} catch (MetaNotFoundException e) {
status.setStatusCode(TStatusCode.NOT_FOUND);
status.addToErrorMsgs(e.getMessage());
} catch (UserException e) {
status.setStatusCode(TStatusCode.INVALID_ARGUMENT);
status.addToErrorMsgs(e.getMessage());
} catch (Exception e) {
LOG.warn("got exception add columns: ", e);
status.setStatusCode(TStatusCode.INTERNAL_ERROR);
status.addToErrorMsgs(e.getMessage());
}
TAddColumnsResult result = new TAddColumnsResult();
result.setStatus(status);
result.setTableId(request.getTableId());
result.setAllColumns(allColumns);
result.setSchemaVersion(schemaVersion);
LOG.debug("result: {}", result);
return result;
}
@LogException
@Override
public TGetTablesResult getTableNames(TGetTablesParams params) throws TException {