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

[feature-wip](multi-catalog) Catalog operation syntax (#10033)

Browse Source 
Impl catalog operation syntax
This commit is contained in:
huangzhaowei
2022-06-17 17:50:31 +08:00
committed by GitHub
parent f35b235c3b
commit 6baa694bc1
32 changed files with 1474 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
fe/fe-core/src/main/cup/sql_parser.cup
View File

@ -277,7 +277,8 @@ terminal String KW_ADD, KW_ADMIN, KW_AFTER, KW_AGGREGATE, KW_ALIAS, KW_ALL, KW_A
KW_VALUE, KW_VALUES, KW_VARCHAR, KW_VARIABLES, KW_VERBOSE, KW_VIEW,
KW_WARNINGS, KW_WEEK, KW_WHEN, KW_WHITELIST, KW_WHERE, KW_WITH, KW_WORK, KW_WRITE,
KW_YEAR,
KW_NOT_NULL;
KW_NOT_NULL,
KW_CATALOG, KW_CATALOGS;
terminal COMMA, COLON, DOT, DOTDOTDOT, AT, STAR, LPAREN, RPAREN, SEMICOLON, LBRACKET, RBRACKET, DIVIDE, MOD, ADD, SUBTRACT;
terminal BITAND, BITOR, BITXOR, BITNOT;
@ -863,6 +864,15 @@ alter_stmt ::=
{:
RESULT = new AlterDatabasePropertyStmt(dbName, map);
:}
/* Catalog */
| KW_ALTER KW_CATALOG ident:catalogName KW_RENAME ident:newCatalogName
{:
RESULT = new AlterCatalogNameStmt(catalogName, newCatalogName);
:}
| KW_ALTER KW_CATALOG ident:catalogName KW_SET KW_PROPERTIES LPAREN key_value_map:map RPAREN
{:
RESULT = new AlterCatalogPropertyStmt(catalogName, map);
:}
| KW_ALTER KW_RESOURCE ident_or_text:resourceName opt_properties:properties
{:
RESULT = new AlterResourceStmt(resourceName, properties);
@ -1225,6 +1235,11 @@ create_stmt ::=
{:
RESULT = new CreateDbStmt(ifNotExists, db, null);
:}
/* Catalog */
| KW_CREATE KW_CATALOG opt_if_not_exists:ifNotExists ident:catalogName opt_properties:properties
{:
RESULT = new CreateCatalogStmt(ifNotExists, catalogName, properties);
:}
/* cluster */
/* KW_CREATE KW_CLUSTER ident:name opt_properties:properties KW_IDENTIFIED KW_BY STRING_LITERAL:password
{:
@ -2014,6 +2029,11 @@ drop_stmt ::=
{:
RESULT = new DropDbStmt(ifExists, db, force);
:}
/* Catalog */
| KW_DROP KW_CATALOG opt_if_exists:ifExists ident:catalogName
{:
RESULT = new DropCatalogStmt(ifExists, catalogName);
:}
/* cluster */
| KW_DROP KW_CLUSTER opt_if_exists:ifExists ident:cluster
{:
@ -2705,6 +2725,16 @@ show_param ::=
{:
RESULT = new ShowDbStmt(parser.wild, parser.where);
:}
/* Catalog */
| KW_CATALOGS
{:
RESULT = new ShowCatalogStmt();
:}
/* show Catalog name */
| KW_CATALOG ident:catalogName
{:
RESULT = new ShowCatalogStmt(catalogName);
:}
/* Dynamic Partition */
| KW_DYNAMIC KW_PARTITION KW_TABLES opt_db:db
{:
@ -5826,6 +5856,10 @@ keyword ::=
{: RESULT = id; :}
| KW_CURRENT_TIMESTAMP:id
{: RESULT = id; :}
| KW_CATALOG:id
{: RESULT = id; :}
| KW_CATALOGS:id
{: RESULT = id; :}
;
// Identifier that contain keyword

91
fe/fe-core/src/main/java/org/apache/doris/analysis/AlterCatalogNameStmt.java Normal file
View File

@ -0,0 +1,91 @@
// 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.doris.analysis;
import org.apache.doris.analysis.CompoundPredicate.Operator;
import org.apache.doris.catalog.Catalog;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
import org.apache.doris.common.FeNameFormat;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.InternalDataSource;
import org.apache.doris.mysql.privilege.PaloPrivilege;
import org.apache.doris.mysql.privilege.PrivBitSet;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.qe.ConnectContext;
import com.google.common.base.Strings;
/**
* Statement for alter the catalog name.
*/
public class AlterCatalogNameStmt extends DdlStmt {
private final String catalogName;
private final String newCatalogName;
public AlterCatalogNameStmt(String catalogName, String newCatalogName) {
this.catalogName = catalogName;
this.newCatalogName = newCatalogName;
}
public String getCatalogName() {
return catalogName;
}
public String getNewCatalogName() {
return newCatalogName;
}
@Override
public void analyze(Analyzer analyzer) throws UserException {
super.analyze(analyzer);
if (!Config.enable_multi_catalog) {
throw new AnalysisException("The multi-catalog feature is still in experiment, and you can enable it "
+ "manually by set fe configuration named `enable_multi_catalog` to be ture.");
}
if (Strings.isNullOrEmpty(catalogName)) {
throw new AnalysisException("Datasource name is not set");
}
if (catalogName.equals(InternalDataSource.INTERNAL_DS_NAME)) {
throw new AnalysisException("Internal catalog can't be alter.");
}
if (!Catalog.getCurrentCatalog().getAuth().checkGlobalPriv(ConnectContext.get(),
PrivPredicate.of(PrivBitSet.of(PaloPrivilege.ADMIN_PRIV, PaloPrivilege.ALTER_PRIV), Operator.OR))) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_DBACCESS_DENIED_ERROR,
analyzer.getQualifiedUser(), catalogName);
}
if (Strings.isNullOrEmpty(newCatalogName)) {
throw new AnalysisException("New catalog name is not set");
}
if (newCatalogName.equals(InternalDataSource.INTERNAL_DS_NAME)) {
throw new AnalysisException("Cannot alter a catalog into a build-in name.");
}
FeNameFormat.checkCommonName("catalog", newCatalogName);
}
@Override
public String toSql() {
return "ALTER CATALOG " + catalogName + " RENAME " + newCatalogName;
}
}

73
fe/fe-core/src/main/java/org/apache/doris/analysis/AlterCatalogPropertyStmt.java Normal file
View File

@ -0,0 +1,73 @@
// 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.doris.analysis;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.FeNameFormat;
import org.apache.doris.common.UserException;
import org.apache.doris.common.util.PrintableMap;
import org.apache.doris.datasource.InternalDataSource;
import com.google.common.base.Strings;
import java.util.Map;
/**
* Statement for alter the catalog property.
*/
public class AlterCatalogPropertyStmt extends DdlStmt {
private final String catalogName;
private final Map<String, String> newProperties;
public AlterCatalogPropertyStmt(String catalogName, Map<String, String> newProperties) {
this.catalogName = catalogName;
this.newProperties = newProperties;
}
public String getCatalogName() {
return catalogName;
}
public Map<String, String> getNewProperties() {
return newProperties;
}
@Override
public void analyze(Analyzer analyzer) throws UserException {
super.analyze(analyzer);
if (!Config.enable_multi_catalog) {
throw new AnalysisException("The multi-catalog feature is still in experiment, and you can enable it "
+ "manually by set fe configuration named `enable_multi_catalog` to be ture.");
}
if (Strings.isNullOrEmpty(catalogName)) {
throw new AnalysisException("Datasource name is not set");
}
if (catalogName.equals(InternalDataSource.INTERNAL_DS_NAME)) {
throw new AnalysisException("Internal catalog can't be alter.");
}
FeNameFormat.checkCatalogProperties(newProperties);
}
@Override
public String toSql() {
return "ALTER CATALOG " + catalogName + " SET PROPERTIES ("
+ new PrintableMap<>(newProperties, "=", true, false, ",") + ")";
}
}

99
fe/fe-core/src/main/java/org/apache/doris/analysis/CreateCatalogStmt.java Normal file
View File

@ -0,0 +1,99 @@
// 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.doris.analysis;
import org.apache.doris.catalog.Catalog;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
import org.apache.doris.common.FeNameFormat;
import org.apache.doris.common.UserException;
import org.apache.doris.common.util.PrintableMap;
import org.apache.doris.datasource.InternalDataSource;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.qe.ConnectContext;
import java.util.HashMap;
import java.util.Map;
/**
* Statement for create a new catalog.
*/
public class CreateCatalogStmt extends DdlStmt {
private final boolean ifNotExists;
private final String catalogName;
private final Map<String, String> properties;
/**
* Statement for create a new catalog.
*/
public CreateCatalogStmt(boolean ifNotExists, String catalogName, Map<String, String> properties) {
this.ifNotExists = ifNotExists;
this.catalogName = catalogName;
this.properties = properties == null ? new HashMap<>() : properties;
}
public String getCatalogName() {
return catalogName;
}
public Map<String, String> getProperties() {
return properties;
}
public boolean isSetIfNotExists() {
return ifNotExists;
}
@Override
public void analyze(Analyzer analyzer) throws UserException {
super.analyze(analyzer);
if (!Config.enable_multi_catalog) {
throw new AnalysisException("The multi-catalog feature is still in experiment, and you can enable it "
+ "manually by set fe configuration named `enable_multi_catalog` to be ture.");
}
if (catalogName.equals(InternalDataSource.INTERNAL_DS_NAME)) {
throw new AnalysisException("Internal catalog name can't be create.");
}
FeNameFormat.checkCommonName("catalog", catalogName);
if (!Catalog.getCurrentCatalog().getAuth().checkGlobalPriv(ConnectContext.get(), PrivPredicate.CREATE)) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_DBACCESS_DENIED_ERROR,
analyzer.getQualifiedUser(), catalogName);
}
FeNameFormat.checkCatalogProperties(properties);
}
@Override
public String toString() {
return toSql();
}
@Override
public String toSql() {
StringBuilder stringBuilder = new StringBuilder();
stringBuilder.append("CREATE CATALOG ").append("`").append(catalogName).append("`");
if (properties.size() > 0) {
stringBuilder.append("\nPROPERTIES (\n");
stringBuilder.append(new PrintableMap<>(properties, "=", true, true, false));
stringBuilder.append("\n)");
}
return stringBuilder.toString();
}
}

82
fe/fe-core/src/main/java/org/apache/doris/analysis/DropCatalogStmt.java Normal file
View File

@ -0,0 +1,82 @@
// 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.doris.analysis;
import org.apache.doris.catalog.Catalog;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.InternalDataSource;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.qe.ConnectContext;
import com.google.common.base.Strings;
/**
* Statement for drop a catalog.
*/
public class DropCatalogStmt extends DdlStmt {
private final boolean ifExists;
private final String catalogName;
public DropCatalogStmt(boolean ifExists, String catalogName) {
this.ifExists = ifExists;
this.catalogName = catalogName;
}
public boolean isSetIfExists() {
return ifExists;
}
public String getCatalogName() {
return this.catalogName;
}
@Override
public void analyze(Analyzer analyzer) throws UserException {
super.analyze(analyzer);
if (!Config.enable_multi_catalog) {
throw new AnalysisException("The multi-catalog feature is still in experiment, and you can enable it "
+ "manually by set fe configuration named `enable_multi_catalog` to be ture.");
}
if (Strings.isNullOrEmpty(catalogName)) {
throw new AnalysisException("Datasource name is not set");
}
if (catalogName.equals(InternalDataSource.INTERNAL_DS_NAME)) {
throw new AnalysisException("Internal catalog can't be drop.");
}
if (!Catalog.getCurrentCatalog().getAuth().checkGlobalPriv(ConnectContext.get(), PrivPredicate.DROP)) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_DBACCESS_DENIED_ERROR,
ConnectContext.get().getQualifiedUser(), catalogName);
}
}
@Override
public String toSql() {
return "DROP CATALOG " + "`" + catalogName + "`";
}
@Override
public String toString() {
return toSql();
}
}

94
fe/fe-core/src/main/java/org/apache/doris/analysis/ShowCatalogStmt.java Normal file
View File

@ -0,0 +1,94 @@
// 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.doris.analysis;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.ScalarType;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.UserException;
import org.apache.doris.qe.ShowResultSetMetaData;
/**
* Statement for show all catalog or desc the specific catalog.
*/
public class ShowCatalogStmt extends ShowStmt {
private static final ShowResultSetMetaData META_DATA_ALL =
ShowResultSetMetaData.builder()
.addColumn(new Column("CatalogName", ScalarType.createVarchar(64)))
.addColumn(new Column("Type", ScalarType.createStringType()))
.build();
private static final ShowResultSetMetaData META_DATA_SPECIFIC =
ShowResultSetMetaData.builder()
.addColumn(new Column("Key", ScalarType.createStringType()))
.addColumn(new Column("Value", ScalarType.createStringType()))
.build();
private final String catalogName;
public ShowCatalogStmt(String catalogName) {
this.catalogName = catalogName;
}
public ShowCatalogStmt() {
this.catalogName = null;
}
public String getCatalogName() {
return catalogName;
}
@Override
public void analyze(Analyzer analyzer) throws AnalysisException, UserException {
if (!Config.enable_multi_catalog) {
throw new AnalysisException("The multi-catalog feature is still in experiment, and you can enable it "
+ "manually by set fe configuration named `enable_multi_catalog` to be ture.");
}
super.analyze(analyzer);
}
@Override
public String toSql() {
StringBuilder sb = new StringBuilder();
sb.append("SHOW");
if (catalogName != null) {
sb.append(" CATALOG ");
sb.append(catalogName);
} else {
sb.append(" CATALOGS");
}
return sb.toString();
}
@Override
public String toString() {
return toSql();
}
@Override
public ShowResultSetMetaData getMetaData() {
if (catalogName == null) {
return META_DATA_ALL;
} else {
return META_DATA_SPECIFIC;
}
}
}

27
fe/fe-core/src/main/java/org/apache/doris/catalog/Catalog.java
View File

@ -1862,6 +1862,22 @@ public class Catalog {
return checksum;
}
/**
* Load datasource through file.
**/
public long loadDatasource(DataInputStream in, long checksum) throws IOException {
if (Config.enable_multi_catalog) {
DataSourceMgr mgr = DataSourceMgr.read(in);
// When enable the multi catalog in the first time, the mgr will be a null value.
// So ignore it to use default datasource manager.
if (mgr != null) {
this.dataSourceMgr = mgr;
}
LOG.info("finished replay datasource from image");
}
return checksum;
}
// Only called by checkpoint thread
// return the latest image file's absolute path
public String saveImage() throws IOException {
@ -2126,6 +2142,17 @@ public class Catalog {
return checksum;
}
/**
* Save datasource image.
*/
public long saveDatasource(CountingDataOutputStream out, long checksum) throws IOException {
// Do not write datasource image when enable multi catalog is false.
if (Config.enable_multi_catalog) {
Catalog.getCurrentCatalog().getDataSourceMgr().write(out);
}
return checksum;
}
public void createLabelCleaner() {
labelCleaner = new MasterDaemon("LoadLabelCleaner", Config.label_clean_interval_second * 1000L) {
@Override

2
fe/fe-core/src/main/java/org/apache/doris/common/ErrorCode.java
View File

@ -1685,7 +1685,7 @@ public enum ErrorCode {
"data cannot be inserted into table with empty partition. "
+ "Use `SHOW PARTITIONS FROM %s` to see the currently partitions of this table. "),
ERROR_SQL_AND_LIMITATIONS_SET_IN_ONE_RULE(5084, new byte[]{'4', '2', '0', '0', '0'},
"sql/sqlHash and partition_num/tablet_num/cardinality cannot be set in one rule.")
"sql/sqlHash and partition_num/tablet_num/cardinality cannot be set in one rule."),
;
// This is error code

11
fe/fe-core/src/main/java/org/apache/doris/common/FeNameFormat.java
View File

@ -23,6 +23,8 @@ import org.apache.doris.system.SystemInfoService;
import com.google.common.base.Strings;
import java.util.Map;
public class FeNameFormat {
private static final String LABEL_REGEX = "^[-_A-Za-z0-9]{1,128}$";
private static final String COMMON_NAME_REGEX = "^[a-zA-Z][a-zA-Z0-9_]{0,63}$";
@ -112,4 +114,13 @@ public class FeNameFormat {
ErrorReport.reportAnalysisException(ErrorCode.ERR_WRONG_NAME_FORMAT, type, name);
}
}
/**
* Check the type property of the catalog props.
*/
public static void checkCatalogProperties(Map<String, String> props) throws AnalysisException {
if (!props.containsKey("type")) {
throw new AnalysisException("All the external catalog should contain the type property.");
}
}
}

82
fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java Normal file
View File

@ -0,0 +1,82 @@
// 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.doris.datasource;
import org.apache.doris.analysis.AlterCatalogNameStmt;
import org.apache.doris.analysis.AlterCatalogPropertyStmt;
import org.apache.doris.analysis.CreateCatalogStmt;
import org.apache.doris.analysis.DropCatalogStmt;
import org.apache.doris.analysis.ShowCatalogStmt;
import org.apache.doris.analysis.StatementBase;
import java.util.Map;
/**
* A factory to create catalog instance of log or covert catalog into log.
*/
public class CatalogFactory {
/**
* Convert the sql statement into catalog log.
*/
public static CatalogLog constructorCatalogLog(StatementBase stmt) {
CatalogLog log = new CatalogLog();
if (stmt instanceof CreateCatalogStmt) {
log.setCatalogName(((CreateCatalogStmt) stmt).getCatalogName());
log.setProps(((CreateCatalogStmt) stmt).getProperties());
} else if (stmt instanceof DropCatalogStmt) {
log.setCatalogName(((DropCatalogStmt) stmt).getCatalogName());
} else if (stmt instanceof AlterCatalogPropertyStmt) {
log.setCatalogName(((AlterCatalogPropertyStmt) stmt).getCatalogName());
log.setNewProps(((AlterCatalogPropertyStmt) stmt).getNewProperties());
} else if (stmt instanceof AlterCatalogNameStmt) {
log.setCatalogName(((AlterCatalogNameStmt) stmt).getCatalogName());
log.setNewCatalogName(((AlterCatalogNameStmt) stmt).getNewCatalogName());
} else if (stmt instanceof ShowCatalogStmt) {
if (((ShowCatalogStmt) stmt).getCatalogName() != null) {
log.setCatalogName(((ShowCatalogStmt) stmt).getCatalogName());
}
} else {
throw new RuntimeException("Unknown stmt for datasource manager " + stmt.getClass().getSimpleName());
}
return log;
}
/**
* create the datasource instance from data source log.
*/
public static DataSourceIf constructorFromLog(CatalogLog log) {
return constructorDataSource(log.getCatalogName(), log.getProps());
}
private static DataSourceIf constructorDataSource(String name, Map<String, String> props) {
String type = props.get("type");
DataSourceIf dataSource;
switch (type) {
case "hms":
dataSource = new HMSExternalDataSource(name, props);
break;
case "es":
dataSource = new EsExternalDataSource(name, props);
break;
default:
throw new RuntimeException("Unknown datasource type for " + type);
}
return dataSource;
}
}

62
fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogLog.java Normal file
View File

@ -0,0 +1,62 @@
// 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.doris.datasource;
import org.apache.doris.common.io.Text;
import org.apache.doris.common.io.Writable;
import org.apache.doris.persist.gson.GsonUtils;
import com.google.gson.annotations.SerializedName;
import lombok.Data;
import lombok.Getter;
import lombok.NoArgsConstructor;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Map;
/**
* A union metadata log for all the catalog operator include create,drop and alter.
*/
@NoArgsConstructor
@Getter
@Data
public class CatalogLog implements Writable {
@SerializedName(value = "catalogName")
private String catalogName;
@SerializedName(value = "props")
private Map<String, String> props;
@SerializedName(value = "newCatalogName")
private String newCatalogName;
@SerializedName(value = "newProps")
private Map<String, String> newProps;
@Override
public void write(DataOutput out) throws IOException {
Text.writeString(out, GsonUtils.GSON.toJson(this));
}
public static CatalogLog read(DataInput in) throws IOException {
String json = Text.readString(in);
return GsonUtils.GSON.fromJson(json, CatalogLog.class);
}
}

9
fe/fe-core/src/main/java/org/apache/doris/datasource/DataSourceIf.java
View File

@ -23,11 +23,12 @@ import org.apache.doris.common.DdlException;
import org.apache.doris.common.MetaNotFoundException;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import javax.annotation.Nullable;
/**
*
* The interface of DataSource(catalog).
*/
public interface DataSourceIf {
@ -66,4 +67,10 @@ public interface DataSourceIf {
DatabaseIf getDbOrAnalysisException(String dbName) throws AnalysisException;
DatabaseIf getDbOrAnalysisException(long dbId) throws AnalysisException;
Map<String, String> getProperties();
void modifyDatasourceName(String name);
void modifyDatasourceProps(Map<String, String> props);
}

266
fe/fe-core/src/main/java/org/apache/doris/datasource/DataSourceMgr.java
View File

@ -17,11 +17,22 @@
package org.apache.doris.datasource;
import org.apache.doris.common.Config;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.analysis.AlterCatalogNameStmt;
import org.apache.doris.analysis.AlterCatalogPropertyStmt;
import org.apache.doris.analysis.CreateCatalogStmt;
import org.apache.doris.analysis.DropCatalogStmt;
import org.apache.doris.analysis.ShowCatalogStmt;
import org.apache.doris.catalog.Catalog;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.UserException;
import org.apache.doris.common.io.Text;
import org.apache.doris.common.io.Writable;
import org.apache.doris.persist.OperationType;
import org.apache.doris.persist.gson.GsonUtils;
import org.apache.doris.qe.ShowResultSet;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@ -29,18 +40,22 @@ import org.apache.logging.log4j.Logger;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
* DataSourceMgr will loaded all data sources at FE startup,
* and save them in maps mapping with id and name.
* DataSourceMgr will load all data sources at FE startup,
* and save them in map with name.
* Note: Catalog in sql syntax will be treated as datasource interface in code level.
* TODO: Change the package name into catalog.
*/
public class DataSourceMgr implements Writable {
private static final Logger LOG = LogManager.getLogger(DataSourceMgr.class);
private Map<Long, DataSourceIf> idToDataSource = Maps.newConcurrentMap();
private Map<String, DataSourceIf> nameToDataSource = Maps.newConcurrentMap();
private DataSourceMgrProperty dsMgrProperty = new DataSourceMgrProperty();
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
private final Map<String, DataSourceIf> nameToCatalogs = Maps.newConcurrentMap();
// Use a separate instance to facilitate access.
// internalDataSource still exists in idToDataSource and nameToDataSource
@ -52,103 +67,182 @@ public class DataSourceMgr implements Writable {
private void initInternalDataSource() {
internalDataSource = new InternalDataSource();
idToDataSource.put(internalDataSource.getId(), internalDataSource);
nameToDataSource.put(internalDataSource.getName(), internalDataSource);
}
private void registerNewDataSource(ExternalDataSource ds) {
// TODO
nameToCatalogs.put(internalDataSource.getName(), internalDataSource);
}
public InternalDataSource getInternalDataSource() {
return internalDataSource;
}
/**
* get data source by id.
*
* @param id
* @param e
* @param <E>
* @return
* @throws E
*/
public <E extends MetaNotFoundException> DataSourceIf getDataSourceOrException(long id,
java.util.function.Function<Long, E> e) throws E {
DataSourceIf ds = idToDataSource.get(id);
if (ds == null) {
throw e.apply(id);
}
return ds;
private void writeLock() {
lock.writeLock().lock();
}
private void writeUnlock() {
lock.writeLock().unlock();
}
private void readLock() {
lock.readLock().lock();
}
private void readUnlock() {
lock.readLock().unlock();
}
/**
* get data source by name.
*
* @param name
* @param e
* @param <E>
* @return
* @throws E
* Create and hold the catalog instance and write the meta log.
*/
public <E extends MetaNotFoundException> DataSourceIf getDataSourceOrException(String name,
java.util.function.Function<String, E> e) throws E {
DataSourceIf ds = nameToDataSource.get(name);
if (ds == null) {
throw e.apply(name);
public void createCatalog(CreateCatalogStmt stmt) throws UserException {
if (stmt.isSetIfNotExists() && nameToCatalogs.containsKey(stmt.getCatalogName())) {
LOG.warn("Catalog {} is already exist.", stmt.getCatalogName());
return;
}
return ds;
if (nameToCatalogs.containsKey(stmt.getCatalogName())) {
throw new DdlException("Catalog had already exist with name: " + stmt.getCatalogName());
}
CatalogLog log = CatalogFactory.constructorCatalogLog(stmt);
replayCreateCatalog(log);
Catalog.getCurrentCatalog().getEditLog().logDatasourceLog(OperationType.OP_CREATE_DS, log);
}
public boolean hasDataSource(String name) {
return nameToDataSource.containsKey(name);
/**
* Remove the catalog instance by name and write the meta log.
*/
public void dropCatalog(DropCatalogStmt stmt) throws UserException {
if (stmt.isSetIfExists() && !nameToCatalogs.containsKey(stmt.getCatalogName())) {
LOG.warn("Non catalog {} is found.", stmt.getCatalogName());
return;
}
if (!nameToCatalogs.containsKey(stmt.getCatalogName())) {
throw new DdlException("No catalog found with name: " + stmt.getCatalogName());
}
CatalogLog log = CatalogFactory.constructorCatalogLog(stmt);
replayDropCatalog(log);
Catalog.getCurrentCatalog().getEditLog().logDatasourceLog(OperationType.OP_DROP_DS, log);
}
/**
* Modify the catalog name into a new one and write the meta log.
*/
public void alterCatalogName(AlterCatalogNameStmt stmt) throws UserException {
if (!nameToCatalogs.containsKey(stmt.getCatalogName())) {
throw new DdlException("No catalog found with name: " + stmt.getCatalogName());
}
CatalogLog log = CatalogFactory.constructorCatalogLog(stmt);
replayAlterCatalogName(log);
Catalog.getCurrentCatalog().getEditLog().logDatasourceLog(OperationType.OP_ALTER_DS_NAME, log);
}
/**
* Modify the catalog property and write the meta log.
*/
public void alterCatalogProps(AlterCatalogPropertyStmt stmt) throws UserException {
if (!nameToCatalogs.containsKey(stmt.getCatalogName())) {
throw new DdlException("No catalog found with name: " + stmt.getCatalogName());
}
if (!nameToCatalogs.get(stmt.getCatalogName())
.getType().equalsIgnoreCase(stmt.getNewProperties().get("type"))) {
throw new DdlException("Can't modify the type of catalog property with name: " + stmt.getCatalogName());
}
CatalogLog log = CatalogFactory.constructorCatalogLog(stmt);
replayAlterCatalogProps(log);
Catalog.getCurrentCatalog().getEditLog().logDatasourceLog(OperationType.OP_ALTER_DS_PROPS, log);
}
/**
* List all catalog or get the special catalog with a name.
*/
public ShowResultSet showCatalogs(ShowCatalogStmt showStmt) throws AnalysisException {
List<List<String>> rows = Lists.newArrayList();
readLock();
try {
if (showStmt.getCatalogName() == null) {
for (DataSourceIf ds : nameToCatalogs.values()) {
List<String> row = Lists.newArrayList();
row.add(ds.getName());
row.add(ds.getType());
rows.add(row);
}
} else {
if (!nameToCatalogs.containsKey(showStmt.getCatalogName())) {
throw new AnalysisException("No catalog found with name: " + showStmt.getCatalogName());
}
DataSourceIf ds = nameToCatalogs.get(showStmt.getCatalogName());
for (Map.Entry<String, String> elem : ds.getProperties().entrySet()) {
List<String> row = Lists.newArrayList();
row.add(elem.getKey());
row.add(elem.getValue());
rows.add(row);
}
}
} finally {
readUnlock();
}
return new ShowResultSet(showStmt.getMetaData(), rows);
}
/**
* Reply for create catalog event.
*/
public void replayCreateCatalog(CatalogLog log) {
writeLock();
try {
DataSourceIf ds = CatalogFactory.constructorFromLog(log);
nameToCatalogs.put(ds.getName(), ds);
} finally {
writeUnlock();
}
}
/**
* Reply for drop catalog event.
*/
public void replayDropCatalog(CatalogLog log) {
writeLock();
try {
nameToCatalogs.remove(log.getCatalogName());
} finally {
writeUnlock();
}
}
/**
* Reply for alter catalog name event.
*/
public void replayAlterCatalogName(CatalogLog log) {
writeLock();
try {
DataSourceIf ds = nameToCatalogs.remove(log.getCatalogName());
ds.modifyDatasourceName(log.getNewCatalogName());
nameToCatalogs.put(ds.getName(), ds);
} finally {
writeUnlock();
}
}
/**
* Reply for alter catalog props event.
*/
public void replayAlterCatalogProps(CatalogLog log) {
writeLock();
try {
DataSourceIf ds = nameToCatalogs.remove(log.getCatalogName());
ds.modifyDatasourceProps(log.getNewProps());
nameToCatalogs.put(ds.getName(), ds);
} finally {
writeUnlock();
}
}
@Override
public void write(DataOutput out) throws IOException {
if (Config.disable_cluster_feature) {
return;
}
Preconditions.checkState(false, "Do not call this until multi catalog feature is ready");
int size = idToDataSource.size();
if (idToDataSource.get(InternalDataSource.INTERNAL_DS_ID) != null) {
// No need to persis internal data source
size -= 1;
}
out.writeInt(size);
for (DataSourceIf ds : idToDataSource.values()) {
if (ds.getId() == InternalDataSource.INTERNAL_DS_ID) {
continue;
}
ExternalDataSource extDs = (ExternalDataSource) ds;
extDs.write(out);
}
dsMgrProperty.write(out);
Text.writeString(out, GsonUtils.GSON.toJson(this));
}
/**
* read from image.
*
* @param in
* @return
* @throws IOException
*/
public static DataSourceMgr read(DataInput in) throws IOException {
if (Config.disable_cluster_feature) {
return null;
}
DataSourceMgr mgr = new DataSourceMgr();
mgr.readFields(in);
return mgr;
}
private void readFields(DataInput in) throws IOException {
int size = in.readInt();
for (int i = 0; i < size; ++i) {
ExternalDataSource extDs = ExternalDataSource.read(in);
idToDataSource.put(extDs.getId(), extDs);
nameToDataSource.put(extDs.getName(), extDs);
}
dsMgrProperty = DataSourceMgrProperty.read(in);
String json = Text.readString(in);
return GsonUtils.GSON.fromJson(json, DataSourceMgr.class);
}
}

5
fe/fe-core/src/main/java/org/apache/doris/datasource/DataSourceProperty.java
View File

@ -23,12 +23,17 @@ import org.apache.doris.persist.gson.GsonUtils;
import com.google.common.collect.Maps;
import com.google.gson.annotations.SerializedName;
import lombok.Data;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Map;
/**
* DataSourceProperty to store the properties for datasource.
*/
@Data
public class DataSourceProperty implements Writable {
@SerializedName(value = "properties")
private Map<String, String> properties = Maps.newHashMap();

10
fe/fe-core/src/main/java/org/apache/doris/datasource/EsExternalDataSource.java
View File

@ -18,12 +18,22 @@
package org.apache.doris.datasource;
import java.util.List;
import java.util.Map;
/**
* External data source for elasticsearch
*/
public class EsExternalDataSource extends ExternalDataSource {
/**
* Default constructor for EsExternalDataSource.
*/
public EsExternalDataSource(String name, Map<String, String> props) {
setName(name);
getDsProperty().setProperties(props);
setType("es");
}
@Override
public List<String> listDatabaseNames(SessionContext ctx) {
return null;

20
fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDataSource.java
View File

@ -17,7 +17,6 @@
package org.apache.doris.datasource;
import org.apache.commons.lang.NotImplementedException;
import org.apache.doris.catalog.DatabaseIf;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.DdlException;
@ -27,18 +26,22 @@ import org.apache.doris.common.io.Writable;
import org.apache.doris.persist.gson.GsonUtils;
import com.google.gson.annotations.SerializedName;
import lombok.Data;
import org.apache.commons.lang.NotImplementedException;
import org.jetbrains.annotations.Nullable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
/**
* The abstract class for all types of external data sources.
*/
@Data
public abstract class ExternalDataSource implements DataSourceIf, Writable {
// Unique id of this data source, will be assigned after data source is loaded.
@SerializedName(value = "id")
@ -153,6 +156,21 @@ public abstract class ExternalDataSource implements DataSourceIf, Writable {
throw new NotImplementedException();
}
@Override
public Map<String, String> getProperties() {
return dsProperty.getProperties();
}
@Override
public void modifyDatasourceName(String name) {
this.name = name;
}
@Override
public void modifyDatasourceProps(Map<String, String> props) {
dsProperty.setProperties(props);
}
@Override
public void write(DataOutput out) throws IOException {
Text.writeString(out, GsonUtils.GSON.toJson(this));

36
fe/fe-core/src/main/java/org/apache/doris/datasource/HMSExternalDataSource.java
View File

@ -53,14 +53,22 @@ public class HMSExternalDataSource extends ExternalDataSource {
protected String hiveMetastoreUris;
protected HiveMetaStoreClient client;
/**
* Default constructor for HMSExternalDataSource.
*/
public HMSExternalDataSource(String name, Map<String, String> props) {
setName(name);
getDsProperty().setProperties(props);
setType("hms");
}
/**
* Hive metastore data source implementation.
*
* @param hiveMetastoreUris e.g. thrift://127.0.0.1:9083
*/
public HMSExternalDataSource(long id, String name, String type, DataSourceProperty dsProperty,
String hiveMetastoreUris)
throws DdlException {
String hiveMetastoreUris) throws DdlException {
this.id = id;
this.name = name;
this.type = type;
@ -186,37 +194,37 @@ public class HMSExternalDataSource extends ExternalDataSource {
@Override
public DatabaseIf getDbOrMetaException(String dbName) throws MetaNotFoundException {
return getDbOrException(dbName, s -> new MetaNotFoundException("unknown databases, dbName=" + s,
ErrorCode.ERR_BAD_DB_ERROR));
return getDbOrException(dbName,
s -> new MetaNotFoundException("unknown databases, dbName=" + s, ErrorCode.ERR_BAD_DB_ERROR));
}
@Override
public DatabaseIf getDbOrMetaException(long dbId) throws MetaNotFoundException {
return getDbOrException(dbId, s -> new MetaNotFoundException("unknown databases, dbId=" + s,
ErrorCode.ERR_BAD_DB_ERROR));
return getDbOrException(dbId,
s -> new MetaNotFoundException("unknown databases, dbId=" + s, ErrorCode.ERR_BAD_DB_ERROR));
}
@Override
public DatabaseIf getDbOrDdlException(String dbName) throws DdlException {
return getDbOrException(dbName, s -> new DdlException(ErrorCode.ERR_BAD_DB_ERROR.formatErrorMsg(s),
ErrorCode.ERR_BAD_DB_ERROR));
return getDbOrException(dbName,
s -> new DdlException(ErrorCode.ERR_BAD_DB_ERROR.formatErrorMsg(s), ErrorCode.ERR_BAD_DB_ERROR));
}
@Override
public DatabaseIf getDbOrDdlException(long dbId) throws DdlException {
return getDbOrException(dbId, s -> new DdlException(ErrorCode.ERR_BAD_DB_ERROR.formatErrorMsg(s),
ErrorCode.ERR_BAD_DB_ERROR));
return getDbOrException(dbId,
s -> new DdlException(ErrorCode.ERR_BAD_DB_ERROR.formatErrorMsg(s), ErrorCode.ERR_BAD_DB_ERROR));
}
@Override
public DatabaseIf getDbOrAnalysisException(String dbName) throws AnalysisException {
return getDbOrException(dbName, s -> new AnalysisException(ErrorCode.ERR_BAD_DB_ERROR.formatErrorMsg(s),
ErrorCode.ERR_BAD_DB_ERROR));
return getDbOrException(dbName,
s -> new AnalysisException(ErrorCode.ERR_BAD_DB_ERROR.formatErrorMsg(s), ErrorCode.ERR_BAD_DB_ERROR));
}
@Override
public DatabaseIf getDbOrAnalysisException(long dbId) throws AnalysisException {
return getDbOrException(dbId, s -> new AnalysisException(ErrorCode.ERR_BAD_DB_ERROR.formatErrorMsg(s),
ErrorCode.ERR_BAD_DB_ERROR));
return getDbOrException(dbId,
s -> new AnalysisException(ErrorCode.ERR_BAD_DB_ERROR.formatErrorMsg(s), ErrorCode.ERR_BAD_DB_ERROR));
}
}

15
fe/fe-core/src/main/java/org/apache/doris/datasource/InternalDataSource.java
View File

@ -326,6 +326,21 @@ public class InternalDataSource implements DataSourceIf {
s -> new AnalysisException(ErrorCode.ERR_BAD_DB_ERROR.formatErrorMsg(s), ErrorCode.ERR_BAD_DB_ERROR));
}
@Override
public Map<String, String> getProperties() {
return Maps.newHashMap();
}
@Override
public void modifyDatasourceName(String name) {
LOG.warn("Ignore the modify datasource name in build-in datasource.");
}
@Override
public void modifyDatasourceProps(Map<String, String> props) {
LOG.warn("Ignore the modify datasource props in build-in datasource.");
}
// Use tryLock to avoid potential dead lock
private boolean tryLock(boolean mustLock) {
while (true) {

9
fe/fe-core/src/main/java/org/apache/doris/journal/JournalEntity.java
View File

@ -36,6 +36,7 @@ import org.apache.doris.cluster.Cluster;
import org.apache.doris.common.io.Text;
import org.apache.doris.common.io.Writable;
import org.apache.doris.common.util.SmallFileMgr.SmallFile;
import org.apache.doris.datasource.CatalogLog;
import org.apache.doris.ha.MasterInfo;
import org.apache.doris.journal.bdbje.Timestamp;
import org.apache.doris.load.DeleteInfo;
@ -653,6 +654,14 @@ public class JournalEntity implements Writable {
isRead = true;
break;
}
case OperationType.OP_CREATE_DS:
case OperationType.OP_DROP_DS:
case OperationType.OP_ALTER_DS_NAME:
case OperationType.OP_ALTER_DS_PROPS: {
data = CatalogLog.read(in);
isRead = true;
break;
}
default: {
IOException e = new IOException();
LOG.error("UNKNOWN Operation Type {}", opCode, e);

25
fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
View File

@ -41,6 +41,7 @@ import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.io.Text;
import org.apache.doris.common.io.Writable;
import org.apache.doris.common.util.SmallFileMgr.SmallFile;
import org.apache.doris.datasource.CatalogLog;
import org.apache.doris.ha.MasterInfo;
import org.apache.doris.journal.Journal;
import org.apache.doris.journal.JournalCursor;
@ -821,6 +822,26 @@ public class EditLog {
catalog.getPolicyMgr().replayDrop(log);
break;
}
case OperationType.OP_CREATE_DS: {
CatalogLog log = (CatalogLog) journal.getData();
catalog.getDataSourceMgr().replayCreateCatalog(log);
break;
}
case OperationType.OP_DROP_DS: {
CatalogLog log = (CatalogLog) journal.getData();
catalog.getDataSourceMgr().replayDropCatalog(log);
break;
}
case OperationType.OP_ALTER_DS_NAME: {
CatalogLog log = (CatalogLog) journal.getData();
catalog.getDataSourceMgr().replayAlterCatalogName(log);
break;
}
case OperationType.OP_ALTER_DS_PROPS: {
CatalogLog log = (CatalogLog) journal.getData();
catalog.getDataSourceMgr().replayAlterCatalogProps(log);
break;
}
default: {
IOException e = new IOException();
LOG.error("UNKNOWN Operation Type {}", opCode, e);
@ -1431,4 +1452,8 @@ public class EditLog {
public void logDropPolicy(DropPolicyLog log) {
logEdit(OperationType.OP_DROP_POLICY, log);
}
public void logDatasourceLog(short id, CatalogLog log) {
logEdit(id, log);
}
}

6
fe/fe-core/src/main/java/org/apache/doris/persist/OperationType.java
View File

@ -223,6 +223,12 @@ public class OperationType {
public static final short OP_CREATE_POLICY = 310;
public static final short OP_DROP_POLICY = 311;
// datasource 312-315
public static final short OP_CREATE_DS = 312;
public static final short OP_DROP_DS = 313;
public static final short OP_ALTER_DS_NAME = 314;
public static final short OP_ALTER_DS_PROPS = 315;
// get opcode name by op codeStri
public static String getOpName(short opCode) {
try {

6
fe/fe-core/src/main/java/org/apache/doris/persist/meta/MetaPersistMethod.java
View File

@ -195,6 +195,12 @@ public class MetaPersistMethod {
metaPersistMethod.writeMethod =
Catalog.class.getDeclaredMethod("savePolicy", CountingDataOutputStream.class, long.class);
break;
case "datasource":
metaPersistMethod.readMethod =
Catalog.class.getDeclaredMethod("loadDatasource", DataInputStream.class, long.class);
metaPersistMethod.writeMethod =
Catalog.class.getDeclaredMethod("saveDatasource", CountingDataOutputStream.class, long.class);
break;
default:
break;
}

2
fe/fe-core/src/main/java/org/apache/doris/persist/meta/PersistMetaModules.java
View File

@ -38,7 +38,7 @@ public class PersistMetaModules {
new String[] {"masterInfo", "frontends", "backends", "db", "loadJob", "alterJob", "recycleBin",
"globalVariable", "cluster", "broker", "resources", "exportJob", "syncJob", "backupHandler",
"paloAuth", "transactionState", "colocateTableIndex", "routineLoadJobs", "loadJobV2", "smallFiles",
"plugins", "deleteHandler", "sqlBlockRule", "policy"});
"plugins", "deleteHandler", "sqlBlockRule", "policy", "datasource"});
static {
MODULES_MAP = Maps.newHashMap();

12
fe/fe-core/src/main/java/org/apache/doris/qe/DdlExecutor.java
View File

@ -26,6 +26,8 @@ import org.apache.doris.analysis.AdminRebalanceDiskStmt;
import org.apache.doris.analysis.AdminRepairTableStmt;
import org.apache.doris.analysis.AdminSetConfigStmt;
import org.apache.doris.analysis.AdminSetReplicaStatusStmt;
import org.apache.doris.analysis.AlterCatalogNameStmt;
import org.apache.doris.analysis.AlterCatalogPropertyStmt;
import org.apache.doris.analysis.AlterClusterStmt;
import org.apache.doris.analysis.AlterColumnStatsStmt;
import org.apache.doris.analysis.AlterDatabasePropertyStmt;
@ -44,6 +46,7 @@ import org.apache.doris.analysis.CancelAlterSystemStmt;
import org.apache.doris.analysis.CancelAlterTableStmt;
import org.apache.doris.analysis.CancelBackupStmt;
import org.apache.doris.analysis.CancelLoadStmt;
import org.apache.doris.analysis.CreateCatalogStmt;
import org.apache.doris.analysis.CreateClusterStmt;
import org.apache.doris.analysis.CreateDataSyncJobStmt;
import org.apache.doris.analysis.CreateDbStmt;
@ -64,6 +67,7 @@ import org.apache.doris.analysis.CreateUserStmt;
import org.apache.doris.analysis.CreateViewStmt;
import org.apache.doris.analysis.DdlStmt;
import org.apache.doris.analysis.DeleteStmt;
import org.apache.doris.analysis.DropCatalogStmt;
import org.apache.doris.analysis.DropClusterStmt;
import org.apache.doris.analysis.DropDbStmt;
import org.apache.doris.analysis.DropEncryptKeyStmt;
@ -300,6 +304,14 @@ public class DdlExecutor {
catalog.getPolicyMgr().createPolicy((CreatePolicyStmt) ddlStmt);
} else if (ddlStmt instanceof DropPolicyStmt) {
catalog.getPolicyMgr().dropPolicy((DropPolicyStmt) ddlStmt);
} else if (ddlStmt instanceof CreateCatalogStmt) {
catalog.getDataSourceMgr().createCatalog((CreateCatalogStmt) ddlStmt);
} else if (ddlStmt instanceof DropCatalogStmt) {
catalog.getDataSourceMgr().dropCatalog((DropCatalogStmt) ddlStmt);
} else if (ddlStmt instanceof AlterCatalogNameStmt) {
catalog.getDataSourceMgr().alterCatalogName((AlterCatalogNameStmt) ddlStmt);
} else if (ddlStmt instanceof AlterCatalogPropertyStmt) {
catalog.getDataSourceMgr().alterCatalogProps((AlterCatalogPropertyStmt) ddlStmt);
} else {
throw new DdlException("Unknown statement.");
}

7
fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java
View File

@ -30,6 +30,7 @@ import org.apache.doris.analysis.ShowAuthorStmt;
import org.apache.doris.analysis.ShowBackendsStmt;
import org.apache.doris.analysis.ShowBackupStmt;
import org.apache.doris.analysis.ShowBrokerStmt;
import org.apache.doris.analysis.ShowCatalogStmt;
import org.apache.doris.analysis.ShowClusterStmt;
import org.apache.doris.analysis.ShowCollationStmt;
import org.apache.doris.analysis.ShowColumnStatsStmt;
@ -346,6 +347,8 @@ public class ShowExecutor {
handleShowCreateMaterializedView();
} else if (stmt instanceof ShowPolicyStmt) {
handleShowPolicy();
} else if (stmt instanceof ShowCatalogStmt) {
handleDatasource();
} else {
handleEmtpy();
}
@ -2211,4 +2214,8 @@ public class ShowExecutor {
resultSet = Catalog.getCurrentCatalog().getPolicyMgr().showPolicy(showStmt);
}
public void handleDatasource() throws AnalysisException {
ShowCatalogStmt showStmt = (ShowCatalogStmt) stmt;
resultSet = Catalog.getCurrentCatalog().getDataSourceMgr().showCatalogs(showStmt);
}
}

2
fe/fe-core/src/main/jflex/sql_scanner.flex
View File

@ -433,6 +433,8 @@ import org.apache.doris.qe.SqlModeHelper;
keywordMap.put("||", new Integer(SqlParserSymbols.KW_PIPE));
keywordMap.put("current_timestamp", new Integer(SqlParserSymbols.KW_CURRENT_TIMESTAMP));
keywordMap.put("not_null", new Integer(SqlParserSymbols.KW_NOT_NULL));
keywordMap.put("catalog", new Integer(SqlParserSymbols.KW_CATALOG));
keywordMap.put("catalogs", new Integer(SqlParserSymbols.KW_CATALOGS));
}
// map from token id to token description

94
fe/fe-core/src/test/java/org/apache/doris/analysis/AlterCatalogNameStmtTest.java Normal file
View File

@ -0,0 +1,94 @@
// 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.doris.analysis;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.InternalDataSource;
import org.apache.doris.mysql.privilege.MockedAuth;
import org.apache.doris.mysql.privilege.PaloAuth;
import org.apache.doris.qe.ConnectContext;
import mockit.Mocked;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
public class AlterCatalogNameStmtTest {
private Analyzer analyzer;
@Mocked
private PaloAuth auth;
@Mocked
private ConnectContext ctx;
@Before
public void setUp() throws DdlException {
Config.enable_multi_catalog = true;
analyzer = AccessTestUtil.fetchAdminAnalyzer(false);
MockedAuth.mockedAuth(auth);
MockedAuth.mockedConnectContext(ctx, "root", "127.0.0.1");
}
@Test
public void testNormalCase() throws UserException {
AlterCatalogNameStmt stmt = new AlterCatalogNameStmt("testCatalog", "testNewCatalog");
stmt.analyze(analyzer);
Assert.assertEquals("testCatalog", stmt.getCatalogName());
Assert.assertEquals("testNewCatalog", stmt.getNewCatalogName());
}
@Test(expected = AnalysisException.class)
public void testEmptyDs1() throws UserException {
AlterCatalogNameStmt stmt = new AlterCatalogNameStmt("", "testNewCatalog");
stmt.analyze(analyzer);
Assert.fail("No exception throws.");
}
@Test(expected = AnalysisException.class)
public void testEmptyDs2() throws UserException {
AlterCatalogNameStmt stmt = new AlterCatalogNameStmt("testCatalog", "");
stmt.analyze(analyzer);
Assert.fail("No exception throws.");
}
@Test(expected = AnalysisException.class)
public void testBuildIn1() throws UserException {
AlterCatalogNameStmt stmt = new AlterCatalogNameStmt(
InternalDataSource.INTERNAL_DS_NAME, "testNewCatalog");
stmt.analyze(analyzer);
Assert.fail("No exception throws.");
}
@Test(expected = AnalysisException.class)
public void testBuildIn2() throws UserException {
AlterCatalogNameStmt stmt = new AlterCatalogNameStmt(
"testCatalog", InternalDataSource.INTERNAL_DS_NAME);
stmt.analyze(analyzer);
Assert.fail("No exception throws.");
}
@Test(expected = AnalysisException.class)
public void testNameFormat() throws UserException {
AlterCatalogNameStmt stmt = new AlterCatalogNameStmt(
"testCatalog", InternalDataSource.INTERNAL_DS_NAME);
stmt.analyze(analyzer);
Assert.fail("No exception throws.");
}
}

91
fe/fe-core/src/test/java/org/apache/doris/analysis/AlterCatalogPropsStmtTest.java Normal file
View File

@ -0,0 +1,91 @@
// 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.doris.analysis;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.InternalDataSource;
import org.apache.doris.mysql.privilege.MockedAuth;
import org.apache.doris.mysql.privilege.PaloAuth;
import org.apache.doris.qe.ConnectContext;
import com.google.common.collect.Maps;
import mockit.Mocked;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import java.util.Map;
public class AlterCatalogPropsStmtTest {
private Analyzer analyzer;
@Mocked
private PaloAuth auth;
@Mocked
private ConnectContext ctx;
@Before
public void setUp() throws DdlException {
Config.enable_multi_catalog = true;
analyzer = AccessTestUtil.fetchAdminAnalyzer(false);
MockedAuth.mockedAuth(auth);
MockedAuth.mockedConnectContext(ctx, "root", "127.0.0.1");
}
@Test
public void testNormalCase() throws UserException {
Map<String, String> props = Maps.newHashMap();
props.put("type", "hms");
props.put("hive.metastore.uris", "thrift://localhost:9083");
AlterCatalogPropertyStmt stmt = new AlterCatalogPropertyStmt("testCatalog", props);
stmt.analyze(analyzer);
Assert.assertEquals("testCatalog", stmt.getCatalogName());
Assert.assertEquals(2, stmt.getNewProperties().size());
}
@Test(expected = AnalysisException.class)
public void testName() throws UserException {
Map<String, String> props = Maps.newHashMap();
props.put("type", "hms");
props.put("hive.metastore.uris", "thrift://localhost:9083");
AlterCatalogPropertyStmt stmt = new AlterCatalogPropertyStmt("", props);
stmt.analyze(analyzer);
Assert.fail("No exception throws.");
}
@Test(expected = AnalysisException.class)
public void testBuildIn() throws UserException {
Map<String, String> props = Maps.newHashMap();
props.put("type", "hms");
props.put("hive.metastore.uris", "thrift://localhost:9083");
AlterCatalogPropertyStmt stmt = new AlterCatalogPropertyStmt(InternalDataSource.INTERNAL_DS_NAME, props);
stmt.analyze(analyzer);
Assert.fail("No exception throws.");
}
@Test(expected = AnalysisException.class)
public void testPropType() throws UserException {
Map<String, String> props = Maps.newHashMap();
props.put("hive.metastore.uris", "thrift://localhost:9083");
AlterCatalogPropertyStmt stmt = new AlterCatalogPropertyStmt(InternalDataSource.INTERNAL_DS_NAME, props);
stmt.analyze(analyzer);
Assert.fail("No exception throws.");
}
}

93
fe/fe-core/src/test/java/org/apache/doris/analysis/CreateCatalogStmtTest.java Normal file
View File

@ -0,0 +1,93 @@
// 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.doris.analysis;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.InternalDataSource;
import org.apache.doris.mysql.privilege.MockedAuth;
import org.apache.doris.mysql.privilege.PaloAuth;
import org.apache.doris.qe.ConnectContext;
import com.google.common.collect.Maps;
import mockit.Mocked;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import java.util.Map;
public class CreateCatalogStmtTest {
private Analyzer analyzer;
@Mocked
private PaloAuth auth;
@Mocked
private ConnectContext ctx;
@Before()
public void setUp() throws DdlException {
Config.enable_multi_catalog = true;
analyzer = AccessTestUtil.fetchAdminAnalyzer(true);
MockedAuth.mockedAuth(auth);
MockedAuth.mockedConnectContext(ctx, "root", "127.0.0.1");
}
@Test
public void testAnalyzeNormal() throws UserException {
Map<String, String> props = Maps.newHashMap();
props.put("type", "hms");
props.put("hive.metastore.uris", "thrift://localhost:9083");
CreateCatalogStmt stmt = new CreateCatalogStmt(false, "testCatalog", props);
stmt.analyze(analyzer);
Assert.assertEquals("testCatalog", stmt.getCatalogName());
Assert.assertNotNull(stmt.getProperties());
Assert.assertEquals(2, stmt.getProperties().size());
}
@Test(expected = AnalysisException.class)
public void testAnalyzeWithException() throws UserException {
Map<String, String> props = Maps.newHashMap();
props.put("type", "hms");
props.put("hive.metastore.uris", "thrift://localhost:9083");
CreateCatalogStmt stmt = new CreateCatalogStmt(false, "", props);
stmt.analyze(analyzer);
Assert.fail("no exception");
}
@Test(expected = AnalysisException.class)
public void testBuildInException() throws UserException {
Map<String, String> props = Maps.newHashMap();
props.put("type", "hms");
props.put("hive.metastore.uris", "thrift://localhost:9083");
CreateCatalogStmt stmt = new CreateCatalogStmt(false, InternalDataSource.INTERNAL_DS_NAME, props);
stmt.analyze(analyzer);
Assert.fail("no exception");
}
@Test(expected = AnalysisException.class)
public void testPropsTypeException() throws UserException {
Map<String, String> props = Maps.newHashMap();
props.put("hive.metastore.uris", "thrift://localhost:9083");
CreateCatalogStmt stmt = new CreateCatalogStmt(false, InternalDataSource.INTERNAL_DS_NAME, props);
stmt.analyze(analyzer);
Assert.fail("no exception");
}
}

72
fe/fe-core/src/test/java/org/apache/doris/analysis/DropCatalogStmtTest.java Normal file
View File

@ -0,0 +1,72 @@
// 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.doris.analysis;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.InternalDataSource;
import org.apache.doris.mysql.privilege.MockedAuth;
import org.apache.doris.mysql.privilege.PaloAuth;
import org.apache.doris.qe.ConnectContext;
import mockit.Mocked;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
public class DropCatalogStmtTest {
Analyzer analyzer;
@Mocked
private PaloAuth auth;
@Mocked
private ConnectContext ctx;
@Before
public void setUp() throws DdlException {
Config.enable_multi_catalog = true;
analyzer = AccessTestUtil.fetchAdminAnalyzer(true);
MockedAuth.mockedAuth(auth);
MockedAuth.mockedConnectContext(ctx, "root", "127.0.0.1");
}
@Test
public void testNormal() throws UserException, AnalysisException {
DropCatalogStmt stmt = new DropCatalogStmt(false, "testCatalog");
stmt.analyze(analyzer);
Assert.assertEquals("testCatalog", stmt.getCatalogName());
}
@Test(expected = AnalysisException.class)
public void testBuildInName() throws UserException, AnalysisException {
DropCatalogStmt stmt = new DropCatalogStmt(false, InternalDataSource.INTERNAL_DS_NAME);
stmt.analyze(analyzer);
Assert.assertEquals("testCatalog", stmt.getCatalogName());
}
@Test(expected = AnalysisException.class)
public void testBadName() throws UserException, AnalysisException {
DropCatalogStmt stmt = new DropCatalogStmt(false, "");
stmt.analyze(analyzer);
Assert.assertEquals("testCatalog", stmt.getCatalogName());
}
}

44
fe/fe-core/src/test/java/org/apache/doris/analysis/ShowCatalogStmtTest.java Normal file
View File

@ -0,0 +1,44 @@
// 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.doris.analysis;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.UserException;
import org.junit.Assert;
import org.junit.Test;
public class ShowCatalogStmtTest {
@Test
public void testNormal() throws UserException, AnalysisException {
Config.enable_multi_catalog = true;
final Analyzer analyzer = AccessTestUtil.fetchBlockAnalyzer();
ShowCatalogStmt stmt = new ShowCatalogStmt();
stmt.analyze(analyzer);
Assert.assertNull(stmt.getCatalogName());
Assert.assertEquals(2, stmt.getMetaData().getColumnCount());
Assert.assertEquals("SHOW CATALOGS", stmt.toSql());
stmt = new ShowCatalogStmt("testCatalog");
stmt.analyze(analyzer);
Assert.assertNotNull(stmt.getCatalogName());
Assert.assertEquals(2, stmt.getMetaData().getColumnCount());
Assert.assertEquals("SHOW CATALOG testCatalog", stmt.toSql());
}
}

96
fe/fe-core/src/test/java/org/apache/doris/datasource/DatasourceMgrTest.java Normal file
View File

@ -0,0 +1,96 @@
// 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.doris.datasource;
import org.apache.doris.analysis.AlterCatalogNameStmt;
import org.apache.doris.analysis.AlterCatalogPropertyStmt;
import org.apache.doris.analysis.CreateCatalogStmt;
import org.apache.doris.analysis.DropCatalogStmt;
import org.apache.doris.analysis.ShowCatalogStmt;
import org.apache.doris.catalog.Catalog;
import org.apache.doris.common.Config;
import org.apache.doris.common.FeConstants;
import org.apache.doris.qe.ShowResultSet;
import org.apache.doris.utframe.TestWithFeService;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.List;
public class DatasourceMgrTest extends TestWithFeService {
private DataSourceMgr mgr;
@Override
protected void runBeforeAll() throws Exception {
Config.enable_multi_catalog = true;
FeConstants.runningUnitTest = true;
mgr = Catalog.getCurrentCatalog().getDataSourceMgr();
}
@Test
public void testNormalCase() throws Exception {
String createCatalogSql = "CREATE CATALOG hms_catalog "
+ "properties( \"type\" = \"hms\", \"hive.metastore.uris\"=\"thrift://localhost:9083\" )";
CreateCatalogStmt createStmt = (CreateCatalogStmt) parseAndAnalyzeStmt(createCatalogSql);
mgr.createCatalog(createStmt);
String showCatalogSql = "SHOW CATALOGS";
ShowCatalogStmt showStmt = (ShowCatalogStmt) parseAndAnalyzeStmt(showCatalogSql);
ShowResultSet showResultSet = mgr.showCatalogs(showStmt);
Assertions.assertEquals(2, showResultSet.getResultRows().size());
String alterCatalogNameSql = "ALTER CATALOG hms_catalog RENAME my_catalog;";
AlterCatalogNameStmt alterNameStmt = (AlterCatalogNameStmt) parseAndAnalyzeStmt(alterCatalogNameSql);
mgr.alterCatalogName(alterNameStmt);
String alterCatalogProps = "ALTER CATALOG my_catalog SET PROPERTIES"
+ " (\"type\" = \"hms\", \"k\" = \"v\");";
AlterCatalogPropertyStmt alterPropStmt = (AlterCatalogPropertyStmt) parseAndAnalyzeStmt(alterCatalogProps);
mgr.alterCatalogProps(alterPropStmt);
showResultSet = mgr.showCatalogs(showStmt);
for (List<String> row : showResultSet.getResultRows()) {
if (row.get(1).equals("internal")) {
continue;
}
Assertions.assertEquals("my_catalog", row.get(0));
}
String showDetailCatalog = "SHOW CATALOG my_catalog";
ShowCatalogStmt showDetailStmt = (ShowCatalogStmt) parseAndAnalyzeStmt(showDetailCatalog);
showResultSet = mgr.showCatalogs(showDetailStmt);
for (List<String> row : showResultSet.getResultRows()) {
Assertions.assertEquals(2, row.size());
if (row.get(0).equalsIgnoreCase("type")) {
Assertions.assertEquals("hms", row.get(1));
} else if (row.get(0).equalsIgnoreCase("k")) {
Assertions.assertEquals("v", row.get(1));
} else {
Assertions.fail();
}
}
String dropCatalogSql = "DROP CATALOG my_catalog";
DropCatalogStmt dropCatalogStmt = (DropCatalogStmt) parseAndAnalyzeStmt(dropCatalogSql);
mgr.dropCatalog(dropCatalogStmt);
showResultSet = mgr.showCatalogs(showStmt);
Assertions.assertEquals(1, showResultSet.getResultRows().size());
}
}