You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by GitBox <gi...@apache.org> on 2022/04/06 16:51:09 UTC
[GitHub] [shardingsphere] terrymanu commented on a diff in pull request #16511: Support multiple data source aggregation for mysql (#15876)
terrymanu commented on code in PR #16511:
URL: https://github.com/apache/shardingsphere/pull/16511#discussion_r844156572
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/config/schema/impl/DataSourceGeneratedSchemaConfiguration.java:
##########
@@ -41,14 +41,27 @@
private final Map<String, DataSource> dataSources;
private final Collection<RuleConfiguration> ruleConfigurations;
+
+ private final Map<String, DataSourceProperties> dataSourceProperties;
Review Comment:
It is better to change `dataSourceProperties` to `dataSourcePropertiesMap`
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/datasource/props/DataSourceProperties.java:
##########
@@ -119,4 +122,84 @@ public int hashCode() {
}
return Objects.hashCode(dataSourceClassName, stringBuilder.toString());
}
+
+ /**
+ * can data source be aggregate.
Review Comment:
First letter of java doc should be upper case
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/datasource/props/DataSourceProperties.java:
##########
@@ -119,4 +122,84 @@ public int hashCode() {
}
return Objects.hashCode(dataSourceClassName, stringBuilder.toString());
}
+
+ /**
+ * can data source be aggregate.
+ *
+ * @param dataSourceProperties data source properties
+ * @throws InvalidDataSourcePropertiesException invalid data source properties exception
+ */
+ public void dataSourceCanBeAggregate(final DataSourceProperties dataSourceProperties) throws InvalidDataSourcePropertiesException {
Review Comment:
How about change method name to `checkToBeAggregatedDataSources` ?
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/config/schema/impl/DataSourceGeneratedSchemaConfiguration.java:
##########
@@ -41,14 +41,27 @@
private final Map<String, DataSource> dataSources;
private final Collection<RuleConfiguration> ruleConfigurations;
+
+ private final Map<String, DataSourceProperties> dataSourceProperties;
public DataSourceGeneratedSchemaConfiguration(final Map<String, DataSourceConfiguration> dataSources, final Collection<RuleConfiguration> ruleConfigurations) {
this.dataSources = DataSourcePoolCreator.create(createDataSourcePropertiesMap(dataSources));
this.ruleConfigurations = ruleConfigurations;
+ this.dataSourceProperties = createDataSourcePropertiesMap(dataSources);
}
private Map<String, DataSourceProperties> createDataSourcePropertiesMap(final Map<String, DataSourceConfiguration> dataSources) {
return dataSources.entrySet().stream().collect(Collectors.toMap(Entry::getKey,
entry -> DataSourcePropertiesCreator.create("com.zaxxer.hikari.HikariDataSource", entry.getValue()), (oldValue, currentValue) -> oldValue, LinkedHashMap::new));
}
+
+ /**
+ * Get data source properties.
+ *
+ * @return data source map
+ */
+ public Map<String, DataSourceProperties> getDataSourceProperties() {
+ return this.dataSourceProperties;
+ }
Review Comment:
Can replace it to lombok's @Getter
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/datasource/props/DataSourceProperties.java:
##########
@@ -119,4 +122,84 @@ public int hashCode() {
}
return Objects.hashCode(dataSourceClassName, stringBuilder.toString());
}
+
+ /**
+ * can data source be aggregate.
+ *
+ * @param dataSourceProperties data source properties
+ * @throws InvalidDataSourcePropertiesException invalid data source properties exception
+ */
+ public void dataSourceCanBeAggregate(final DataSourceProperties dataSourceProperties) throws InvalidDataSourcePropertiesException {
+ if (!parseJdbcUrl().contains("jdbc:mysql")) {
+ return;
+ }
+ for (Entry<String, Object> entry : getAllLocalProperties().entrySet()) {
+ if (!dataSourceProperties.getAllLocalProperties().containsKey(entry.getKey())) {
+ String msg = "target data source item:" + entry.getKey() + ", not in source data source:" + dataSourceProperties.getAllLocalProperties();
Review Comment:
":" should be keep a space with after word
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/datasource/registry/GlobalDataSourceRegistry.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.shardingsphere.infra.datasource.registry;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import javax.sql.DataSource;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+@Setter
+@Getter
+public final class GlobalDataSourceRegistry {
+
+ private static final GlobalDataSourceRegistry INSTANCE = new GlobalDataSourceRegistry();
+
+ private volatile Map<Instance, DataSource> dataSourceCache = new LinkedHashMap<>();
+
+ private volatile Map<String, String> dataSourceSchema = new LinkedHashMap<>();
+
+ private boolean dataSourceAggregationEnabled;
+
+ /**
+ * Get global data source.
+ *
+ * @return instance of ShardingSphere global data source.
+ */
+ public static GlobalDataSourceRegistry getInstance() {
+ return INSTANCE;
+ }
+
Review Comment:
Please remove unnecessary blank line
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/datasource/registry/GlobalDataSourceRegistry.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.shardingsphere.infra.datasource.registry;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import javax.sql.DataSource;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+@Setter
+@Getter
+public final class GlobalDataSourceRegistry {
+
+ private static final GlobalDataSourceRegistry INSTANCE = new GlobalDataSourceRegistry();
+
+ private volatile Map<Instance, DataSource> dataSourceCache = new LinkedHashMap<>();
Review Comment:
How about rename `dataSourceCache` to `cachedDataSources`
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/datasource/registry/Instance.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.shardingsphere.infra.datasource.registry;
+
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+
+/**
+ * instance.
+ */
+@RequiredArgsConstructor
+@Getter
+@EqualsAndHashCode
+public final class Instance {
+
+ private final String ip;
+
+ private final Integer port;
+
+}
+
Review Comment:
Please remove unnecessary blank line
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/datasource/registry/GlobalDataSourceRegistry.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.shardingsphere.infra.datasource.registry;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import javax.sql.DataSource;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+@Setter
+@Getter
+public final class GlobalDataSourceRegistry {
+
+ private static final GlobalDataSourceRegistry INSTANCE = new GlobalDataSourceRegistry();
+
+ private volatile Map<Instance, DataSource> dataSourceCache = new LinkedHashMap<>();
+
+ private volatile Map<String, String> dataSourceSchema = new LinkedHashMap<>();
+
+ private boolean dataSourceAggregationEnabled;
Review Comment:
Missing volatile?
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/datasource/registry/Instance.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.shardingsphere.infra.datasource.registry;
+
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+
+/**
+ * instance.
+ */
+@RequiredArgsConstructor
+@Getter
+@EqualsAndHashCode
+public final class Instance {
+
+ private final String ip;
+
+ private final Integer port;
+
Review Comment:
Please remove unnecessary blank line
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/datasource/registry/GlobalDataSourceRegistry.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.shardingsphere.infra.datasource.registry;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import javax.sql.DataSource;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+@Setter
+@Getter
+public final class GlobalDataSourceRegistry {
+
+ private static final GlobalDataSourceRegistry INSTANCE = new GlobalDataSourceRegistry();
+
+ private volatile Map<Instance, DataSource> dataSourceCache = new LinkedHashMap<>();
+
+ private volatile Map<String, String> dataSourceSchema = new LinkedHashMap<>();
Review Comment:
Is there need to setter?
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/config/yaml/swapper/YamlProxyConfigurationSwapper.java:
##########
@@ -53,21 +53,24 @@ public ProxyConfiguration swap(final YamlProxyConfiguration yamlConfig) {
yamlConfig.getServerConfiguration().getProps(), yamlConfig.getServerConfiguration().getLabels());
return new ProxyConfiguration(schemaConfigs, globalConfig);
}
-
+
private Map<String, DataSourceGeneratedSchemaConfiguration> swapSchemaConfigurations(final YamlProxyConfiguration yamlConfig) {
Map<String, DataSourceGeneratedSchemaConfiguration> result = new LinkedHashMap<>(yamlConfig.getSchemaConfigurations().size(), 1);
+ boolean isDataSourceAggregation = yamlConfig.getServerConfiguration().getProps().containsKey("data-source-aggregation-enabled")
+ && (boolean) yamlConfig.getServerConfiguration().getProps().get("data-source-aggregation-enabled");
for (Entry<String, YamlProxySchemaConfiguration> entry : yamlConfig.getSchemaConfigurations().entrySet()) {
- Map<String, DataSourceConfiguration> schemaDataSourceConfigs = swapDataSourceConfigurations(entry.getValue().getDataSources());
+ Map<String, DataSourceConfiguration> schemaDataSourceConfigs = swapDataSourceConfigurations(entry.getValue().getDataSources(), entry.getValue().getSchemaName(), isDataSourceAggregation);
Collection<RuleConfiguration> schemaRuleConfigs = ruleConfigSwapperEngine.swapToRuleConfigurations(entry.getValue().getRules());
result.put(entry.getKey(), new DataSourceGeneratedSchemaConfiguration(schemaDataSourceConfigs, schemaRuleConfigs));
}
return result;
}
-
- private Map<String, DataSourceConfiguration> swapDataSourceConfigurations(final Map<String, YamlProxyDataSourceConfiguration> yamlConfigs) {
+
Review Comment:
Please keep original indent
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/config/yaml/swapper/YamlProxyConfigurationSwapper.java:
##########
@@ -36,11 +36,11 @@
* YAML proxy configuration swapper.
*/
public final class YamlProxyConfigurationSwapper {
-
+
private final YamlProxyDataSourceConfigurationSwapper dataSourceConfigSwapper = new YamlProxyDataSourceConfigurationSwapper();
-
+
private final YamlRuleConfigurationSwapperEngine ruleConfigSwapperEngine = new YamlRuleConfigurationSwapperEngine();
-
+
Review Comment:
Please keep original indent
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/communication/jdbc/datasource/JDBCBackendDataSource.java:
##########
@@ -100,7 +102,14 @@
private Connection createConnection(final String schemaName, final String dataSourceName, final DataSource dataSource, final TransactionType transactionType) throws SQLException {
ShardingSphereTransactionManager transactionManager
= ProxyContext.getInstance().getContextManager().getTransactionContexts().getEngines().get(schemaName).getTransactionManager(transactionType);
- return isInTransaction(transactionManager) ? transactionManager.getConnection(dataSourceName) : dataSource.getConnection();
+ boolean isDataSourceAggregation = ProxyContext.getInstance().getContextManager().getMetaDataContexts().getProps().getProps().containsKey("data-source-aggregation-enabled")
+ ? ProxyContext.getInstance().getContextManager().getMetaDataContexts().getProps().getValue(ConfigurationPropertyKey.DATA_SOURCE_AGGREGATION_ENABLED) : false;
+ Connection connection = isInTransaction(transactionManager) ? transactionManager.getConnection(dataSourceName) : dataSource.getConnection();
+ if (isDataSourceAggregation) {
+ String databaseName = GlobalDataSourceRegistry.getInstance().getDataSourceSchema().get(schemaName + "." + dataSourceName);
+ connection.setCatalog(databaseName);
+ }
+ return connection;
Review Comment:
`connection` should rename to `result`
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/updatable/ImportSchemaConfigurationHandler.java:
##########
@@ -139,13 +140,13 @@ private void alterRulesConfig(final String schemaName, final Collection<YamlRule
Optional<MetaDataPersistService> metaDataPersistService = metaDataContexts.getMetaDataPersistService();
metaDataPersistService.ifPresent(op -> op.getSchemaRuleService().persist(schemaName, toBeUpdatedRuleConfigs));
}
-
+
private void checkSchemaName(final String schemaName) {
if (!ProxyContext.getInstance().getAllSchemaNames().contains(schemaName)) {
throw new SchemaNotExistedException(schemaName);
}
}
-
+
Review Comment:
Please keep original indent
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/updatable/ImportSchemaConfigurationHandler.java:
##########
@@ -101,7 +102,7 @@ private void alterResourcesConfig(final String schemaName, final Map<String, Yam
throw new InvalidResourcesException(toBeUpdatedResourcePropsMap.keySet());
}
}
-
+
Review Comment:
Please keep original indent
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/test/java/org/apache/shardingsphere/proxy/backend/config/yaml/swapper/YamlProxyDataSourceConfigurationSwapperTest.java:
##########
@@ -33,24 +33,24 @@
import static org.junit.Assert.assertTrue;
public final class YamlProxyDataSourceConfigurationSwapperTest {
-
+
@Test
public void assertSwap() throws IOException {
YamlProxyConfiguration yamlProxyConfig = ProxyConfigurationLoader.load("/conf/swap");
YamlProxyDataSourceConfiguration yamlProxyDataSourceConfig = yamlProxyConfig.getSchemaConfigurations().get("swapper_test").getDataSources().get("foo_db");
- DataSourceConfiguration actualDataSourceConfig = new YamlProxyDataSourceConfigurationSwapper().swap(yamlProxyDataSourceConfig);
+ DataSourceConfiguration actualDataSourceConfig = new YamlProxyDataSourceConfigurationSwapper().swap(yamlProxyDataSourceConfig, "swapper_test", "foo_db", false);
assertConnectionConfig(actualDataSourceConfig);
assertPoolConfig(actualDataSourceConfig);
}
-
+
private void assertConnectionConfig(final DataSourceConfiguration actualDataSourceConfig) {
ConnectionConfiguration actualConnection = actualDataSourceConfig.getConnection();
assertNotNull(actualConnection);
assertThat(actualConnection.getUrl(), is("jdbc:h2:mem:foo_db;DB_CLOSE_DELAY=-1"));
assertThat(actualConnection.getUsername(), is("sa"));
assertThat(actualConnection.getPassword(), is(""));
}
-
+
Review Comment:
Please keep original indent
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/updatable/ImportSchemaConfigurationHandler.java:
##########
@@ -139,13 +140,13 @@ private void alterRulesConfig(final String schemaName, final Collection<YamlRule
Optional<MetaDataPersistService> metaDataPersistService = metaDataContexts.getMetaDataPersistService();
metaDataPersistService.ifPresent(op -> op.getSchemaRuleService().persist(schemaName, toBeUpdatedRuleConfigs));
}
-
+
Review Comment:
Please keep original indent
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/updatable/ImportSchemaConfigurationHandler.java:
##########
@@ -74,19 +74,20 @@
* Import schema configuration handler.
*/
public final class ImportSchemaConfigurationHandler extends UpdatableRALBackendHandler<ImportSchemaConfigurationStatement, ImportSchemaConfigurationHandler> {
-
+
private final DataSourcePropertiesValidator validator = new DataSourcePropertiesValidator();
-
+
private final ShardingRuleConfigurationImportChecker shardingRuleConfigurationImportChecker = new ShardingRuleConfigurationImportChecker();
private final ReadwriteSplittingRuleConfigurationImportChecker readwriteSplittingRuleConfigurationImportChecker = new ReadwriteSplittingRuleConfigurationImportChecker();
private final YamlProxyDataSourceConfigurationSwapper dataSourceConfigSwapper = new YamlProxyDataSourceConfigurationSwapper();
-
+
Review Comment:
Please keep original indent
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/datasource/props/DataSourceProperties.java:
##########
@@ -119,4 +122,84 @@ public int hashCode() {
}
return Objects.hashCode(dataSourceClassName, stringBuilder.toString());
}
+
+ /**
+ * can data source be aggregate.
+ *
+ * @param dataSourceProperties data source properties
+ * @throws InvalidDataSourcePropertiesException invalid data source properties exception
+ */
+ public void dataSourceCanBeAggregate(final DataSourceProperties dataSourceProperties) throws InvalidDataSourcePropertiesException {
+ if (!parseJdbcUrl().contains("jdbc:mysql")) {
+ return;
+ }
+ for (Entry<String, Object> entry : getAllLocalProperties().entrySet()) {
+ if (!dataSourceProperties.getAllLocalProperties().containsKey(entry.getKey())) {
+ String msg = "target data source item:" + entry.getKey() + ", not in source data source:" + dataSourceProperties.getAllLocalProperties();
+ throw new InvalidDataSourcePropertiesException(dataSourceProperties.getDataSourceClassName(), msg);
+ }
+ if (entry.getKey().equals("jdbcUrl") && !isInSameDatabaseInstance(String.valueOf(entry.getValue()),
+ String.valueOf(dataSourceProperties.getAllLocalProperties().get(entry.getKey())))) {
+ String msg = "target data source item:" + entry.getValue() + ", not equal source in data source:" + dataSourceProperties.getAllLocalProperties().get(entry.getKey());
+ throw new InvalidDataSourcePropertiesException(dataSourceProperties.getDataSourceClassName(), msg);
+ }
+ if (!entry.getKey().equals("jdbcUrl") && !String.valueOf(entry.getValue()).equals(String.valueOf(dataSourceProperties.getAllLocalProperties().get(entry.getKey())))) {
+ String msg = "target data source item: " + entry.getKey() + ", value:" + entry.getValue() + ", not equal source item: " + entry.getKey() + ",data source:"
+ + dataSourceProperties.getAllLocalProperties().get(entry.getKey());
+ throw new InvalidDataSourcePropertiesException(dataSourceProperties.getDataSourceClassName(), msg);
+ }
+ }
+ }
+
+ /**
+ * Is data source in same instance.
+ *
+ * @param sourceUrl source url
+ * @param targetUrl target url
+ * @return boolean
+ */
+ public boolean isInSameDatabaseInstance(final String sourceUrl, final String targetUrl) {
+ if (sourceUrl.equals(targetUrl)) {
+ return true;
+ }
+ JdbcUrl sourceJdbcUrl = parseStandardJdbcUrlParserByJdbcUrl(sourceUrl);
+ JdbcUrl targetJdbcUrl = parseStandardJdbcUrlParserByJdbcUrl(targetUrl);
+ return sourceJdbcUrl.getHostname().equals(targetJdbcUrl.getHostname()) && sourceJdbcUrl.getPort() == targetJdbcUrl.getPort()
+ && sourceJdbcUrl.getQueryProperties().equals(targetJdbcUrl.getQueryProperties());
+ }
+
+ private JdbcUrl parseStandardJdbcUrlParserByJdbcUrl(final String jdbcUrl) {
+ StandardJdbcUrlParser jdbcUrlParser = new StandardJdbcUrlParser();
+ return jdbcUrlParser.parse(jdbcUrl);
+ }
+
+ private String parseJdbcUrl() {
+ for (Entry<String, Object> entry : getAllLocalProperties().entrySet()) {
+ if (entry.getKey().equals("jdbcUrl")) {
+ return String.valueOf(entry.getValue());
+ }
+ }
+ return "";
+ }
+
+ /**
+ * get instance from data source.
+ *
+ * @return Instance
+ */
+ public Instance getInstance() {
+ JdbcUrl jdbcUrl = parseStandardJdbcUrlParserByJdbcUrl(parseJdbcUrl());
+ return new Instance(jdbcUrl.getHostname(), jdbcUrl.getPort());
+ }
+
+ /**
+ * get database from data source.
Review Comment:
First letter of java doc should be upper case
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/config/schema/impl/DataSourceGeneratedSchemaConfiguration.java:
##########
@@ -41,14 +41,27 @@
private final Map<String, DataSource> dataSources;
private final Collection<RuleConfiguration> ruleConfigurations;
+
+ private final Map<String, DataSourceProperties> dataSourceProperties;
public DataSourceGeneratedSchemaConfiguration(final Map<String, DataSourceConfiguration> dataSources, final Collection<RuleConfiguration> ruleConfigurations) {
this.dataSources = DataSourcePoolCreator.create(createDataSourcePropertiesMap(dataSources));
this.ruleConfigurations = ruleConfigurations;
+ this.dataSourceProperties = createDataSourcePropertiesMap(dataSources);
Review Comment:
`this.dataSourceProperties` can be change to `dataSourceProperties`, `this.` is unnecessary here
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/datasource/registry/GlobalDataSourceRegistry.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.shardingsphere.infra.datasource.registry;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import javax.sql.DataSource;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+@Setter
+@Getter
+public final class GlobalDataSourceRegistry {
Review Comment:
Missing java doc
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/datasource/props/DataSourceProperties.java:
##########
@@ -119,4 +122,84 @@ public int hashCode() {
}
return Objects.hashCode(dataSourceClassName, stringBuilder.toString());
}
+
+ /**
+ * can data source be aggregate.
+ *
+ * @param dataSourceProperties data source properties
+ * @throws InvalidDataSourcePropertiesException invalid data source properties exception
+ */
+ public void dataSourceCanBeAggregate(final DataSourceProperties dataSourceProperties) throws InvalidDataSourcePropertiesException {
+ if (!parseJdbcUrl().contains("jdbc:mysql")) {
+ return;
+ }
+ for (Entry<String, Object> entry : getAllLocalProperties().entrySet()) {
+ if (!dataSourceProperties.getAllLocalProperties().containsKey(entry.getKey())) {
+ String msg = "target data source item:" + entry.getKey() + ", not in source data source:" + dataSourceProperties.getAllLocalProperties();
+ throw new InvalidDataSourcePropertiesException(dataSourceProperties.getDataSourceClassName(), msg);
+ }
+ if (entry.getKey().equals("jdbcUrl") && !isInSameDatabaseInstance(String.valueOf(entry.getValue()),
+ String.valueOf(dataSourceProperties.getAllLocalProperties().get(entry.getKey())))) {
+ String msg = "target data source item:" + entry.getValue() + ", not equal source in data source:" + dataSourceProperties.getAllLocalProperties().get(entry.getKey());
+ throw new InvalidDataSourcePropertiesException(dataSourceProperties.getDataSourceClassName(), msg);
+ }
+ if (!entry.getKey().equals("jdbcUrl") && !String.valueOf(entry.getValue()).equals(String.valueOf(dataSourceProperties.getAllLocalProperties().get(entry.getKey())))) {
+ String msg = "target data source item: " + entry.getKey() + ", value:" + entry.getValue() + ", not equal source item: " + entry.getKey() + ",data source:"
+ + dataSourceProperties.getAllLocalProperties().get(entry.getKey());
+ throw new InvalidDataSourcePropertiesException(dataSourceProperties.getDataSourceClassName(), msg);
+ }
+ }
+ }
+
+ /**
+ * Is data source in same instance.
+ *
+ * @param sourceUrl source url
+ * @param targetUrl target url
+ * @return boolean
+ */
+ public boolean isInSameDatabaseInstance(final String sourceUrl, final String targetUrl) {
+ if (sourceUrl.equals(targetUrl)) {
+ return true;
+ }
+ JdbcUrl sourceJdbcUrl = parseStandardJdbcUrlParserByJdbcUrl(sourceUrl);
+ JdbcUrl targetJdbcUrl = parseStandardJdbcUrlParserByJdbcUrl(targetUrl);
+ return sourceJdbcUrl.getHostname().equals(targetJdbcUrl.getHostname()) && sourceJdbcUrl.getPort() == targetJdbcUrl.getPort()
+ && sourceJdbcUrl.getQueryProperties().equals(targetJdbcUrl.getQueryProperties());
+ }
+
+ private JdbcUrl parseStandardJdbcUrlParserByJdbcUrl(final String jdbcUrl) {
+ StandardJdbcUrlParser jdbcUrlParser = new StandardJdbcUrlParser();
+ return jdbcUrlParser.parse(jdbcUrl);
+ }
+
+ private String parseJdbcUrl() {
+ for (Entry<String, Object> entry : getAllLocalProperties().entrySet()) {
+ if (entry.getKey().equals("jdbcUrl")) {
+ return String.valueOf(entry.getValue());
+ }
+ }
+ return "";
+ }
+
+ /**
+ * get instance from data source.
Review Comment:
First letter of java doc should be upper case
##########
shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-spi/src/test/java/org/apache/shardingsphere/data/pipeline/api/datasource/config/impl/ShardingSpherePipelineDataSourceConfigurationTest.java:
##########
@@ -58,4 +60,28 @@ private String getDataSourceYaml() {
yamlDataSourceConfigs.forEach((key, value) -> result.put(key, new YamlDataSourceConfigurationSwapper().swapToDataSourceProperties(value)));
return result;
}
+
+ @Test
+ public void assertDataSourceCanBeAggregation() throws InvalidDataSourcePropertiesException {
+ ShardingSpherePipelineDataSourceConfiguration dataSourceConfig = new ShardingSpherePipelineDataSourceConfiguration(getDataSourceAggregationYaml());
+ List<DataSourceProperties> actual = new ArrayList<>(getDataSourcePropertiesMap(dataSourceConfig.getRootConfig().getDataSources()).values());
Review Comment:
How about change ArrayList to LinkedList?
##########
shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/rule/builder/schema/SchemaRulesBuilder.java:
##########
@@ -59,7 +64,13 @@
@SuppressWarnings({"unchecked", "rawtypes"})
public static Collection<ShardingSphereRule> buildRules(final String schemaName, final SchemaConfiguration schemaConfig, final ConfigurationProperties props) {
Collection<ShardingSphereRule> result = new LinkedList<>();
+ boolean isDataSourceAggregation = props.getProps().containsKey("data-source-aggregation-enabled") ? props.getValue(ConfigurationPropertyKey.DATA_SOURCE_AGGREGATION_ENABLED) : false;
for (Entry<RuleConfiguration, SchemaRuleBuilder> entry : getRuleBuilderMap(schemaConfig).entrySet()) {
+ if (isDataSourceAggregation) {
+ DataSourceGeneratedSchemaConfiguration configuration = (DataSourceGeneratedSchemaConfiguration) schemaConfig;
+ checkDataSourceAggregations(configuration.getDataSourceProperties());
+ }
+
Review Comment:
Please remove unnecessary blank line
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/config/yaml/swapper/YamlProxyConfigurationSwapper.java:
##########
@@ -36,11 +36,11 @@
* YAML proxy configuration swapper.
*/
public final class YamlProxyConfigurationSwapper {
-
+
private final YamlProxyDataSourceConfigurationSwapper dataSourceConfigSwapper = new YamlProxyDataSourceConfigurationSwapper();
-
+
Review Comment:
Please keep original indent
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/config/yaml/swapper/YamlProxyConfigurationSwapper.java:
##########
@@ -36,11 +36,11 @@
* YAML proxy configuration swapper.
*/
public final class YamlProxyConfigurationSwapper {
-
+
Review Comment:
Please keep original indent
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/config/yaml/swapper/YamlProxyDataSourceConfigurationSwapper.java:
##########
@@ -31,9 +33,15 @@
* Swap YAML proxy data source configuration to data source configuration.
*
* @param yamlConfig YAML proxy data source configuration
+ * @param schemaName Schema name
+ * @param dataSourceName Data source name
+ * @param isDataSourceAggregation Is data source aggregation
Review Comment:
`Is` should be `is`
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/config/yaml/swapper/YamlProxyConfigurationSwapper.java:
##########
@@ -53,21 +53,24 @@ public ProxyConfiguration swap(final YamlProxyConfiguration yamlConfig) {
yamlConfig.getServerConfiguration().getProps(), yamlConfig.getServerConfiguration().getLabels());
return new ProxyConfiguration(schemaConfigs, globalConfig);
}
-
+
Review Comment:
Please keep original indent
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/updatable/ImportSchemaConfigurationHandler.java:
##########
@@ -74,19 +74,20 @@
* Import schema configuration handler.
*/
public final class ImportSchemaConfigurationHandler extends UpdatableRALBackendHandler<ImportSchemaConfigurationStatement, ImportSchemaConfigurationHandler> {
-
+
Review Comment:
Please keep original indent
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/distsql/ral/common/updatable/ImportSchemaConfigurationHandler.java:
##########
@@ -74,19 +74,20 @@
* Import schema configuration handler.
*/
public final class ImportSchemaConfigurationHandler extends UpdatableRALBackendHandler<ImportSchemaConfigurationStatement, ImportSchemaConfigurationHandler> {
-
+
private final DataSourcePropertiesValidator validator = new DataSourcePropertiesValidator();
-
+
Review Comment:
Please keep original indent
##########
shardingsphere-proxy/shardingsphere-proxy-bootstrap/src/main/resources/conf/server.yaml:
##########
@@ -81,3 +81,4 @@
# sql-federation-enabled: false
# # Available proxy backend driver type: JDBC (default), ExperimentalVertx
# proxy-backend-driver-type: JDBC
+# data-source-aggregation-enabled: false
Review Comment:
Please keep a blank line in the end of line.
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/test/java/org/apache/shardingsphere/proxy/backend/config/yaml/swapper/YamlProxyDataSourceConfigurationSwapperTest.java:
##########
@@ -33,24 +33,24 @@
import static org.junit.Assert.assertTrue;
public final class YamlProxyDataSourceConfigurationSwapperTest {
-
+
@Test
public void assertSwap() throws IOException {
YamlProxyConfiguration yamlProxyConfig = ProxyConfigurationLoader.load("/conf/swap");
YamlProxyDataSourceConfiguration yamlProxyDataSourceConfig = yamlProxyConfig.getSchemaConfigurations().get("swapper_test").getDataSources().get("foo_db");
- DataSourceConfiguration actualDataSourceConfig = new YamlProxyDataSourceConfigurationSwapper().swap(yamlProxyDataSourceConfig);
+ DataSourceConfiguration actualDataSourceConfig = new YamlProxyDataSourceConfigurationSwapper().swap(yamlProxyDataSourceConfig, "swapper_test", "foo_db", false);
assertConnectionConfig(actualDataSourceConfig);
assertPoolConfig(actualDataSourceConfig);
}
-
+
Review Comment:
Please keep original indent
##########
shardingsphere-proxy/shardingsphere-proxy-backend/src/test/java/org/apache/shardingsphere/proxy/backend/config/yaml/swapper/YamlProxyDataSourceConfigurationSwapperTest.java:
##########
@@ -33,24 +33,24 @@
import static org.junit.Assert.assertTrue;
public final class YamlProxyDataSourceConfigurationSwapperTest {
-
+
Review Comment:
Please keep original indent
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@shardingsphere.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org