You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by pa...@apache.org on 2022/11/17 09:45:21 UTC
[shardingsphere] branch master updated: Optimize pg_class, pg_namespace collect logic, add decorate logic. (#22235)
This is an automated email from the ASF dual-hosted git repository.
panjuan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new 10aef772053 Optimize pg_class, pg_namespace collect logic, add decorate logic. (#22235)
10aef772053 is described below
commit 10aef772053b2d2911ae845a5a621ca46bca747f
Author: Chuxin Chen <ch...@qq.com>
AuthorDate: Thu Nov 17 17:45:14 2022 +0800
Optimize pg_class, pg_namespace collect logic, add decorate logic. (#22235)
---
.../data/ShardingStatisticsTableCollector.java | 2 +-
.../collector/tables/PgClassTableCollector.java | 101 ++++++++++++++++-----
.../tables/PgNamespaceTableCollector.java | 21 +++--
.../YamlShardingSphereTableDataSwapper.java | 3 +-
.../ShardingSphereDataScheduleCollector.java | 8 +-
.../ShardingSphereDataCollectorFixture.java | 2 +-
test/integration-test/discovery/pom.xml | 2 +-
7 files changed, 96 insertions(+), 43 deletions(-)
diff --git a/features/sharding/core/src/main/java/org/apache/shardingsphere/sharding/metadata/data/ShardingStatisticsTableCollector.java b/features/sharding/core/src/main/java/org/apache/shardingsphere/sharding/metadata/data/ShardingStatisticsTableCollector.java
index d6bd8453386..b35dd9618e4 100644
--- a/features/sharding/core/src/main/java/org/apache/shardingsphere/sharding/metadata/data/ShardingStatisticsTableCollector.java
+++ b/features/sharding/core/src/main/java/org/apache/shardingsphere/sharding/metadata/data/ShardingStatisticsTableCollector.java
@@ -52,7 +52,7 @@ public final class ShardingStatisticsTableCollector implements ShardingSphereDat
private static final String MYSQL_TABLE_ROWS_AND_DATA_LENGTH = "SELECT TABLE_ROWS, DATA_LENGTH FROM information_schema.TABLES WHERE TABLE_SCHEMA = '%s' AND TABLE_NAME = '%s'";
@Override
- public Optional<ShardingSphereTableData> collect(final String databaseName, final ShardingSphereTable table,
+ public Optional<ShardingSphereTableData> collect(final String databaseName, final ShardingSphereTable table,
final Map<String, ShardingSphereDatabase> shardingSphereDatabases) throws SQLException {
ShardingSphereDatabase shardingSphereDatabase = shardingSphereDatabases.get(databaseName);
Optional<ShardingRule> shardingRule = shardingSphereDatabase.getRuleMetaData().findSingleRule(ShardingRule.class);
diff --git a/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/data/collector/tables/PgClassTableCollector.java b/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/data/collector/tables/PgClassTableCollector.java
index 7da2f8844e3..bb0f39053ef 100644
--- a/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/data/collector/tables/PgClassTableCollector.java
+++ b/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/data/collector/tables/PgClassTableCollector.java
@@ -21,20 +21,25 @@ import org.apache.shardingsphere.infra.metadata.data.ShardingSphereRowData;
import org.apache.shardingsphere.infra.metadata.data.ShardingSphereTableData;
import org.apache.shardingsphere.infra.metadata.data.collector.ShardingSphereDataCollector;
import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
+import org.apache.shardingsphere.infra.metadata.database.schema.decorator.model.ShardingSphereColumn;
import org.apache.shardingsphere.infra.metadata.database.schema.decorator.model.ShardingSphereTable;
+import org.apache.shardingsphere.infra.rule.ShardingSphereRule;
+import org.apache.shardingsphere.infra.rule.identifier.type.DataNodeContainedRule;
import javax.sql.DataSource;
-import java.math.BigInteger;
import java.sql.Connection;
import java.sql.ResultSet;
-import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.stream.Collectors;
/**
* Table pg_class data collector.
@@ -43,44 +48,90 @@ public final class PgClassTableCollector implements ShardingSphereDataCollector
private static final String PG_CLASS = "pg_class";
- private static final String SELECT_SQL = "SELECT oid, relname, relnamespace, reltype, reloftype, relowner, relam, relfilenode, reltablespace, relpages, reltuples, relallvisible, reltoastrelid,"
- + "relhasindex, relisshared, relpersistence, relkind, relnatts, relchecks, relhasrules, relhastriggers, relhassubclass, relrowsecurity, relforcerowsecurity, relispopulated, relreplident,"
- + "relispartition, relrewrite, relfrozenxid, relminmxid, relacl, reloptions, relpartbound FROM pg_catalog.pg_class "
- + "WHERE relkind IN ('r','v','m','S','L','f','e','o','') AND relname NOT LIKE 'matviewmap\\_%' AND relname NOT LIKE 'mlog\\_%' "
- + "AND pg_catalog.pg_table_is_visible(oid);";
+ private static final String COLUMN_NAMES = "relname, relnamespace, relkind";
+
+ private static final String SELECT_SQL = "SELECT " + COLUMN_NAMES + " FROM pg_catalog.pg_class WHERE relkind IN ('r','v','m','S','L','f','e','o','') "
+ + "AND relname NOT LIKE 'matviewmap\\_%' AND relname NOT LIKE 'mlog\\_%' AND pg_catalog.pg_table_is_visible(oid);";
@Override
- public Optional<ShardingSphereTableData> collect(final String databaseName, final ShardingSphereTable table,
+ public Optional<ShardingSphereTableData> collect(final String databaseName, final ShardingSphereTable table,
final Map<String, ShardingSphereDatabase> shardingSphereDatabases) throws SQLException {
- ShardingSphereTableData result = new ShardingSphereTableData(PG_CLASS, new ArrayList<>(table.getColumns().values()));
+ List<ShardingSphereRowData> rows = new LinkedList<>();
for (DataSource each : shardingSphereDatabases.get(databaseName).getResourceMetaData().getDataSources().values()) {
try (
Connection connection = each.getConnection();
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(SELECT_SQL)) {
- ResultSetMetaData metaData = resultSet.getMetaData();
- while (resultSet.next()) {
- List<Object> rows = new ArrayList<>(metaData.getColumnCount());
- for (int i = 1; i <= metaData.getColumnCount(); i++) {
- rows.add(convertIfNecessary(resultSet.getObject(i), metaData.getColumnType(i)));
- }
- ShardingSphereRowData rowData = new ShardingSphereRowData(rows);
- result.getRows().add(rowData);
- }
+ rows.addAll(getRows(resultSet, table));
}
}
+ List<ShardingSphereRowData> rowData = decorateTableName(rows, table, shardingSphereDatabases.get(databaseName).getRuleMetaData().getRules());
+ ShardingSphereTableData result = new ShardingSphereTableData(PG_CLASS, new ArrayList<>(table.getColumns().values()));
+ result.getRows().addAll(rowData.stream().distinct().collect(Collectors.toList()));
return Optional.of(result);
}
- // TODO extract to util
- private Object convertIfNecessary(final Object data, final int dataType) {
- if (Types.ARRAY == dataType || Types.OTHER == dataType) {
- return null == data ? null : data.toString();
+ private List<ShardingSphereRowData> getRows(final ResultSet resultSet, final ShardingSphereTable table) throws SQLException {
+ List<ShardingSphereRowData> result = new LinkedList<>();
+ List<String> selectedColumnNames = Arrays.stream(COLUMN_NAMES.split(",")).map(String::trim).collect(Collectors.toList());
+ while (resultSet.next()) {
+ result.add(new ShardingSphereRowData(getRow(table, resultSet, selectedColumnNames)));
+ }
+ return result;
+ }
+
+ private List<Object> getRow(final ShardingSphereTable table, final ResultSet resultSet, final List<String> selectedColumnNames) throws SQLException {
+ List<Object> result = new LinkedList<>();
+ for (ShardingSphereColumn each : table.getColumns().values()) {
+ if (selectedColumnNames.contains(each.getName())) {
+ result.add(resultSet.getObject(each.getName()));
+ } else {
+ result.add(mockValue(each.getDataType()));
+ }
+ }
+ return result;
+ }
+
+ private Object mockValue(final int dataType) {
+ switch (dataType) {
+ case Types.BIGINT:
+ return 0L;
+ case Types.VARCHAR:
+ case Types.CHAR:
+ case Types.OTHER:
+ case Types.ARRAY:
+ return "";
+ case Types.INTEGER:
+ case Types.SMALLINT:
+ return 0;
+ case Types.REAL:
+ return Float.valueOf("0");
+ case Types.BIT:
+ return false;
+ default:
+ return null;
}
- if (Types.BIGINT == dataType) {
- return null == data ? null : new BigInteger(data.toString());
+ }
+
+ private List<ShardingSphereRowData> decorateTableName(final List<ShardingSphereRowData> rows, final ShardingSphereTable table, final Collection<ShardingSphereRule> rules) {
+ Optional<DataNodeContainedRule> dataNodeContainedRule = rules.stream().filter(rule -> rule instanceof DataNodeContainedRule).map(rule -> (DataNodeContainedRule) rule).findFirst();
+ if (!dataNodeContainedRule.isPresent()) {
+ return rows;
+ }
+ int tableNameIndex = table.getColumnNames().indexOf("relname");
+ List<ShardingSphereRowData> result = new LinkedList<>();
+ for (ShardingSphereRowData each : rows) {
+ String tableName = (String) each.getRows().get(tableNameIndex);
+ Optional<String> logicTableName = dataNodeContainedRule.get().findLogicTableByActualTable(tableName);
+ if (logicTableName.isPresent()) {
+ List<Object> decoratedRow = new ArrayList<>(each.getRows());
+ decoratedRow.set(tableNameIndex, logicTableName.get());
+ result.add(new ShardingSphereRowData(decoratedRow));
+ } else {
+ result.add(each);
+ }
}
- return data;
+ return result;
}
@Override
diff --git a/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/data/collector/tables/PgNamespaceTableCollector.java b/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/data/collector/tables/PgNamespaceTableCollector.java
index c42abf2c732..ead5956a951 100644
--- a/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/data/collector/tables/PgNamespaceTableCollector.java
+++ b/infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/data/collector/tables/PgNamespaceTableCollector.java
@@ -24,7 +24,6 @@ import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.infra.metadata.database.schema.decorator.model.ShardingSphereTable;
import javax.sql.DataSource;
-import java.math.BigInteger;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
@@ -32,9 +31,11 @@ import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
/**
* Table pg_namespace data collector.
@@ -44,9 +45,9 @@ public final class PgNamespaceTableCollector implements ShardingSphereDataCollec
private static final String PG_NAMESPACE = "pg_namespace";
@Override
- public Optional<ShardingSphereTableData> collect(final String databaseName, final ShardingSphereTable table,
+ public Optional<ShardingSphereTableData> collect(final String databaseName, final ShardingSphereTable table,
final Map<String, ShardingSphereDatabase> shardingSphereDatabases) throws SQLException {
- ShardingSphereTableData result = new ShardingSphereTableData(PG_NAMESPACE, new ArrayList<>(table.getColumns().values()));
+ Set<ShardingSphereRowData> rows = new LinkedHashSet<>();
for (DataSource each : shardingSphereDatabases.get(databaseName).getResourceMetaData().getDataSources().values()) {
try (
Connection connection = each.getConnection();
@@ -54,25 +55,27 @@ public final class PgNamespaceTableCollector implements ShardingSphereDataCollec
ResultSet resultSet = statement.executeQuery("SELECT oid, nspname, nspowner, nspacl FROM pg_catalog.pg_namespace")) {
ResultSetMetaData metaData = resultSet.getMetaData();
while (resultSet.next()) {
- List<Object> rows = new ArrayList<>(metaData.getColumnCount());
+ List<Object> row = new ArrayList<>(metaData.getColumnCount());
for (int i = 1; i <= metaData.getColumnCount(); i++) {
- rows.add(convertIfNecessary(resultSet.getObject(i), metaData.getColumnType(i)));
+ row.add(convertIfNecessary(resultSet.getObject(i), metaData.getColumnType(i)));
}
- ShardingSphereRowData rowData = new ShardingSphereRowData(rows);
- result.getRows().add(rowData);
+ ShardingSphereRowData rowData = new ShardingSphereRowData(row);
+ rows.add(rowData);
}
}
}
+ ShardingSphereTableData result = new ShardingSphereTableData(PG_NAMESPACE, new ArrayList<>(table.getColumns().values()));
+ result.getRows().addAll(rows);
return Optional.of(result);
}
// TODO extract to util
private Object convertIfNecessary(final Object data, final int dataType) {
- if (Types.ARRAY == dataType || Types.OTHER == dataType) {
+ if (Types.ARRAY == dataType) {
return null == data ? null : data.toString();
}
if (Types.BIGINT == dataType) {
- return null == data ? null : new BigInteger(data.toString());
+ return null == data ? null : Long.valueOf(data.toString());
}
return data;
}
diff --git a/infra/common/src/main/java/org/apache/shardingsphere/infra/yaml/data/swapper/YamlShardingSphereTableDataSwapper.java b/infra/common/src/main/java/org/apache/shardingsphere/infra/yaml/data/swapper/YamlShardingSphereTableDataSwapper.java
index e104a7b3784..d5d1cdb75bc 100644
--- a/infra/common/src/main/java/org/apache/shardingsphere/infra/yaml/data/swapper/YamlShardingSphereTableDataSwapper.java
+++ b/infra/common/src/main/java/org/apache/shardingsphere/infra/yaml/data/swapper/YamlShardingSphereTableDataSwapper.java
@@ -26,7 +26,6 @@ import org.apache.shardingsphere.infra.yaml.data.pojo.YamlShardingSphereTableDat
import org.apache.shardingsphere.infra.yaml.schema.pojo.YamlShardingSphereColumn;
import java.math.BigDecimal;
-import java.math.BigInteger;
import java.sql.Types;
import java.util.Collections;
import java.util.LinkedList;
@@ -113,7 +112,7 @@ public final class YamlShardingSphereTableDataSwapper implements YamlConfigurati
return new BigDecimal(data.toString());
}
if (Types.BIGINT == dataType) {
- return new BigInteger(data.toString());
+ return Long.valueOf(data.toString());
}
if (Types.REAL == dataType || Types.FLOAT == dataType) {
return Float.parseFloat(data.toString());
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/ShardingSphereDataScheduleCollector.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/ShardingSphereDataScheduleCollector.java
index 1dd845777f2..8ffd1cbb744 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/ShardingSphereDataScheduleCollector.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/ShardingSphereDataScheduleCollector.java
@@ -71,18 +71,18 @@ public final class ShardingSphereDataScheduleCollector {
compareUpdateAndSendEvent(shardingSphereData, changedShardingSphereData);
}
- private void collectForDatabase(final String databaseName, final ShardingSphereDatabaseData databaseData,
+ private void collectForDatabase(final String databaseName, final ShardingSphereDatabaseData databaseData,
final Map<String, ShardingSphereDatabase> databases, final ShardingSphereData changedShardingSphereData) {
databaseData.getSchemaData().forEach((key, value) -> collectForSchema(databaseName, key, value, databases, changedShardingSphereData));
}
- private void collectForSchema(final String databaseName, final String schemaName, final ShardingSphereSchemaData schemaData,
+ private void collectForSchema(final String databaseName, final String schemaName, final ShardingSphereSchemaData schemaData,
final Map<String, ShardingSphereDatabase> databases, final ShardingSphereData changedShardingSphereData) {
- schemaData.getTableData().forEach((key, value) -> collectForTable(databaseName, schemaName, databases.get(databaseName).getSchema(schemaName).getTable(key),
+ schemaData.getTableData().forEach((key, value) -> collectForTable(databaseName, schemaName, databases.get(databaseName).getSchema(schemaName).getTable(key),
databases, changedShardingSphereData));
}
- private void collectForTable(final String databaseName, final String schemaName, final ShardingSphereTable table,
+ private void collectForTable(final String databaseName, final String schemaName, final ShardingSphereTable table,
final Map<String, ShardingSphereDatabase> databases, final ShardingSphereData changedShardingSphereData) {
Optional<ShardingSphereDataCollector> shardingSphereDataCollector = ShardingSphereDataCollectorFactory.findInstance(table.getName());
if (!shardingSphereDataCollector.isPresent()) {
diff --git a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/execute/ShardingSphereDataCollectorFixture.java b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/execute/ShardingSphereDataCollectorFixture.java
index 6efde01e3d8..ada2332e888 100644
--- a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/execute/ShardingSphereDataCollectorFixture.java
+++ b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/execute/ShardingSphereDataCollectorFixture.java
@@ -35,7 +35,7 @@ import java.util.Optional;
public final class ShardingSphereDataCollectorFixture implements ShardingSphereDataCollector {
@Override
- public Optional<ShardingSphereTableData> collect(final String databaseName, final ShardingSphereTable table,
+ public Optional<ShardingSphereTableData> collect(final String databaseName, final ShardingSphereTable table,
final Map<String, ShardingSphereDatabase> shardingSphereDatabases) throws SQLException {
ShardingSphereTableData shardingSphereTableData = new ShardingSphereTableData("test_table", Collections.emptyList());
shardingSphereTableData.getRows().add(new ShardingSphereRowData(Arrays.asList("1", "2")));
diff --git a/test/integration-test/discovery/pom.xml b/test/integration-test/discovery/pom.xml
index bfff7b9d32a..fb358612bed 100644
--- a/test/integration-test/discovery/pom.xml
+++ b/test/integration-test/discovery/pom.xml
@@ -40,7 +40,7 @@
<version>${project.version}</version>
<type>test-jar</type>
</dependency>
-
+
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>