Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support distributed transactions across multiple logical database(#19… #20114

Merged
merged 8 commits into from
Aug 16, 2022
Merged

Support distributed transactions across multiple logical database(#19… #20114

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
c73ce2b
Support distributed transactions across multiple logical database(#19…
FlyingZC Aug 12, 2022
c8d3ddc
Fix test case
FlyingZC Aug 12, 2022
51c4f71
Generate data source name
FlyingZC Aug 13, 2022
7f583aa
Add test case
FlyingZC Aug 13, 2022
d0ca227
Remove final
FlyingZC Aug 13, 2022
adf2f84
JDBC does not support operations across multiple logical databases in…
FlyingZC Aug 15, 2022
8d78d74
Merge remote-tracking branch 'apache/master' into feature-multi-logic…
FlyingZC Aug 15, 2022
7528131
Fix equals usage
FlyingZC Aug 15, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ...ava/org/apache/shardingsphere/infra/merge/engine/merger/impl/TransparentResultMerger.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
public final class TransparentResultMerger implements ResultMerger {

@Override
public MergedResult merge(final List<QueryResult> queryResults, final SQLStatementContext<?> sqlStatementContext,
public MergedResult merge(final List<QueryResult> queryResults, final SQLStatementContext<?> sqlStatementContext,
final ShardingSphereDatabase database, final ConnectionContext connectionContext) {
return new TransparentMergedResult(queryResults.get(0));
}
Expand Down
5 changes: 4 additions & 1 deletion ...ction-core/src/main/java/org/apache/shardingsphere/transaction/ConnectionTransaction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@ public final class ConnectionTransaction {

private final TransactionType transactionType;

private final String databaseName;

@Setter
@Getter
private volatile boolean rollbackOnly;
Expand All @@ -46,6 +48,7 @@ public ConnectionTransaction(final String databaseName, final TransactionRule ru
}

public ConnectionTransaction(final String databaseName, final TransactionType transactionType, final TransactionRule rule) {
this.databaseName = databaseName;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should databaseName be extracted from SQL?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I will check.

this.transactionType = transactionType;
transactionManager = rule.getResource().getTransactionManager(transactionType);
TransactionTypeHolder.set(transactionType);
Expand Down Expand Up @@ -87,7 +90,7 @@ public boolean isHoldTransaction(final boolean autoCommit) {
* @throws SQLException SQL exception
*/
public Optional<Connection> getConnection(final String dataSourceName) throws SQLException {
return isInTransaction() ? Optional.of(transactionManager.getConnection(dataSourceName)) : Optional.empty();
return isInTransaction() ? Optional.of(transactionManager.getConnection(this.databaseName, dataSourceName)) : Optional.empty();
}

/**
Expand Down
5 changes: 4 additions & 1 deletion ...action-core/src/main/java/org/apache/shardingsphere/transaction/rule/TransactionRule.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,10 @@ private synchronized ShardingSphereTransactionManagerEngine createTransactionMan
Map<String, DataSource> dataSourceMap = new HashMap<>(databases.size());
Set<DatabaseType> databaseTypes = new HashSet<>();
for (Entry<String, ShardingSphereDatabase> entry : databases.entrySet()) {
dataSourceMap.putAll(entry.getValue().getResource().getDataSources());
ShardingSphereDatabase database = entry.getValue();
database.getResource().getDataSources().forEach((key, value) -> {
dataSourceMap.put(database.getName() + "." + key, value);
});
if (null != entry.getValue().getResource().getDatabaseType()) {
databaseTypes.add(entry.getValue().getResource().getDatabaseType());
}
Expand Down
3 changes: 2 additions & 1 deletion ...main/java/org/apache/shardingsphere/transaction/spi/ShardingSphereTransactionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,11 +56,12 @@ public interface ShardingSphereTransactionManager extends AutoCloseable {
/**
* Get transactional connection.
*
* @param databaseName database name
* @param dataSourceName data source name
* @return connection
* @throws SQLException SQL exception
*/
Connection getConnection(String dataSourceName) throws SQLException;
Connection getConnection(String databaseName, String dataSourceName) throws SQLException;

/**
* Begin transaction.
Expand Down
2 changes: 1 addition & 1 deletion ...shardingsphere/transaction/core/fixture/OtherShardingSphereTransactionManagerFixture.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ public boolean isInTransaction() {
}

@Override
public Connection getConnection(final String dataSourceName) {
public Connection getConnection(final String databaseName, final String dataSourceName) {
return null;
}

Expand Down
2 changes: 1 addition & 1 deletion ...ache/shardingsphere/transaction/core/fixture/ShardingSphereTransactionManagerFixture.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ public boolean isInTransaction() {
}

@Override
public Connection getConnection(final String dataSourceName) {
public Connection getConnection(final String databaseName, final String dataSourceName) {
return null;
}

Expand Down
4 changes: 2 additions & 2 deletions ...che/shardingsphere/transaction/base/seata/at/SeataATShardingSphereTransactionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,9 +90,9 @@ public boolean isInTransaction() {
}

@Override
public Connection getConnection(final String dataSourceName) throws SQLException {
public Connection getConnection(final String databaseName, final String dataSourceName) throws SQLException {
Preconditions.checkState(enableSeataAT, "sharding seata-at transaction has been disabled.");
return dataSourceMap.get(dataSourceName).getConnection();
return dataSourceMap.get(databaseName + "." + dataSourceName).getConnection();
}

@Override
Expand Down
8 changes: 5 additions & 3 deletions ...shardingsphere/transaction/base/seata/at/SeataATShardingSphereTransactionManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,8 @@ public final class SeataATShardingSphereTransactionManagerTest {

private static final MockSeataServer MOCK_SEATA_SERVER = new MockSeataServer();

private static final String DATA_SOURCE_UNIQUE_NAME = "sharding_db.foo_ds";

private final SeataATShardingSphereTransactionManager seataTransactionManager = new SeataATShardingSphereTransactionManager();

private final Queue<Object> requestQueue = MOCK_SEATA_SERVER.getMessageHandler().getRequestQueue();
Expand All @@ -84,7 +86,7 @@ public static void after() {

@Before
public void setUp() {
seataTransactionManager.init(DatabaseTypeFactory.getInstance("MySQL"), Collections.singletonList(new ResourceDataSource("foo_ds", new MockedDataSource())), "Seata");
seataTransactionManager.init(DatabaseTypeFactory.getInstance("MySQL"), Collections.singletonList(new ResourceDataSource(DATA_SOURCE_UNIQUE_NAME, new MockedDataSource())), "Seata");
}

@After
Expand All @@ -102,13 +104,13 @@ public void tearDown() {
public void assertInit() {
Map<String, DataSource> actual = getDataSourceMap();
assertThat(actual.size(), is(1));
assertThat(actual.get("foo_ds"), instanceOf(DataSourceProxy.class));
assertThat(actual.get(DATA_SOURCE_UNIQUE_NAME), instanceOf(DataSourceProxy.class));
assertThat(seataTransactionManager.getTransactionType(), is(TransactionType.BASE));
}

@Test
public void assertGetConnection() throws SQLException {
Connection actual = seataTransactionManager.getConnection("foo_ds");
Connection actual = seataTransactionManager.getConnection("sharding_db", "foo_ds");
assertThat(actual, instanceOf(ConnectionProxy.class));
}

Expand Down
4 changes: 2 additions & 2 deletions ...ain/java/org/apache/shardingsphere/transaction/xa/XAShardingSphereTransactionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,9 +74,9 @@ public boolean isInTransaction() {
}

@Override
public Connection getConnection(final String dataSourceName) throws SQLException {
public Connection getConnection(final String databaseName, final String dataSourceName) throws SQLException {
try {
return cachedDataSources.get(dataSourceName).getConnection();
return cachedDataSources.get(databaseName + "." + dataSourceName).getConnection();
} catch (final SystemException | RollbackException ex) {
throw new SQLException(ex);
}
Expand Down
18 changes: 9 additions & 9 deletions ...java/org/apache/shardingsphere/transaction/xa/XAShardingSphereTransactionManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,9 +82,9 @@ public void assertIsInTransaction() {
@Test
public void assertGetConnection() throws SQLException {
xaTransactionManager.begin();
Connection actual1 = xaTransactionManager.getConnection("ds1");
Connection actual2 = xaTransactionManager.getConnection("ds2");
Connection actual3 = xaTransactionManager.getConnection("ds3");
Connection actual1 = xaTransactionManager.getConnection("demo_ds_1", "ds1");
Connection actual2 = xaTransactionManager.getConnection("demo_ds_2", "ds2");
Connection actual3 = xaTransactionManager.getConnection("demo_ds_3", "ds3");
assertThat(actual1, instanceOf(Connection.class));
assertThat(actual2, instanceOf(Connection.class));
assertThat(actual3, instanceOf(Connection.class));
Expand All @@ -93,10 +93,10 @@ public void assertGetConnection() throws SQLException {

@Test
public void assertGetConnectionOfNestedTransaction() throws SQLException {
ThreadLocal<Map<Transaction, Connection>> transactions = getEnlistedTransactions(getCachedDataSources().get("ds1"));
ThreadLocal<Map<Transaction, Connection>> transactions = getEnlistedTransactions(getCachedDataSources().get("demo_ds_1.ds1"));
xaTransactionManager.begin();
assertTrue(transactions.get().isEmpty());
xaTransactionManager.getConnection("ds1");
xaTransactionManager.getConnection("demo_ds_1", "ds1");
assertThat(transactions.get().size(), is(1));
executeNestedTransaction(transactions);
assertThat(transactions.get().size(), is(1));
Expand All @@ -106,7 +106,7 @@ public void assertGetConnectionOfNestedTransaction() throws SQLException {

private void executeNestedTransaction(final ThreadLocal<Map<Transaction, Connection>> transactions) throws SQLException {
xaTransactionManager.begin();
xaTransactionManager.getConnection("ds1");
xaTransactionManager.getConnection("demo_ds_1", "ds1");
assertThat(transactions.get().size(), is(2));
xaTransactionManager.commit(false);
assertThat(transactions.get().size(), is(1));
Expand Down Expand Up @@ -153,9 +153,9 @@ private ThreadLocal<Map<Transaction, Connection>> getEnlistedTransactions(final

private Collection<ResourceDataSource> createResourceDataSources(final DatabaseType databaseType) {
List<ResourceDataSource> result = new LinkedList<>();
result.add(new ResourceDataSource("ds1", DataSourceUtils.build(HikariDataSource.class, databaseType, "demo_ds_1")));
result.add(new ResourceDataSource("ds2", DataSourceUtils.build(HikariDataSource.class, databaseType, "demo_ds_2")));
result.add(new ResourceDataSource("ds3", DataSourceUtils.build(AtomikosDataSourceBean.class, databaseType, "demo_ds_3")));
result.add(new ResourceDataSource("demo_ds_1.ds1", DataSourceUtils.build(HikariDataSource.class, databaseType, "demo_ds_1")));
result.add(new ResourceDataSource("demo_ds_2.ds2", DataSourceUtils.build(HikariDataSource.class, databaseType, "demo_ds_2")));
result.add(new ResourceDataSource("demo_ds_3.ds3", DataSourceUtils.build(AtomikosDataSourceBean.class, databaseType, "demo_ds_3")));
return result;
}
}
2 changes: 1 addition & 1 deletion ...che/shardingsphere/proxy/backend/communication/jdbc/datasource/JDBCBackendDataSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,7 @@ private List<Connection> createConnections(final String databaseName, final Stri
private Connection createConnection(final String databaseName, final String dataSourceName, final DataSource dataSource, final TransactionType transactionType) throws SQLException {
TransactionRule transactionRule = ProxyContext.getInstance().getContextManager().getMetaDataContexts().getMetaData().getGlobalRuleMetaData().getSingleRule(TransactionRule.class);
ShardingSphereTransactionManager transactionManager = transactionRule.getResource().getTransactionManager(transactionType);
Connection result = isInTransaction(transactionManager) ? transactionManager.getConnection(dataSourceName) : dataSource.getConnection();
Connection result = isInTransaction(transactionManager) ? transactionManager.getConnection(databaseName, dataSourceName) : dataSource.getConnection();
if (dataSourceName.contains(".")) {
String catalog = dataSourceName.split("\\.")[1];
result.setCatalog(catalog);
Expand Down
4 changes: 0 additions & 4 deletions ...kend/src/main/java/org/apache/shardingsphere/proxy/backend/session/ConnectionSession.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import lombok.Setter;
import org.apache.shardingsphere.infra.config.props.ConfigurationPropertyKey;
import org.apache.shardingsphere.infra.database.type.DatabaseType;
import org.apache.shardingsphere.infra.exception.ShardingSphereException;
import org.apache.shardingsphere.infra.executor.sql.prepare.driver.ExecutorStatementManager;
import org.apache.shardingsphere.infra.metadata.user.Grantee;
import org.apache.shardingsphere.infra.session.ConnectionContext;
Expand Down Expand Up @@ -104,9 +103,6 @@ public void setCurrentDatabase(final String databaseName) {
if (null != databaseName && databaseName.equals(this.databaseName)) {
return;
}
if (transactionStatus.isInTransaction()) {
throw new ShardingSphereException("Failed to switch database, please terminate current transaction.");
}
if (statementManager instanceof JDBCBackendStatement) {
((JDBCBackendStatement) statementManager).setDatabaseName(databaseName);
}
Expand Down
4 changes: 2 additions & 2 deletions .../src/test/java/org/apache/shardingsphere/proxy/backend/session/ConnectionSessionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,8 +78,8 @@ public void assertFailedSwitchTransactionTypeWhileBegin() throws SQLException {
connectionSession.getTransactionStatus().setTransactionType(TransactionType.XA);
}

@Test(expected = ShardingSphereException.class)
public void assertFailedSwitchSchemaWhileBegin() throws SQLException {
@Test
public void assertSwitchSchemaWhileBegin() throws SQLException {
connectionSession.setCurrentDatabase("db");
JDBCBackendTransactionManager transactionManager = new JDBCBackendTransactionManager(backendConnection);
transactionManager.begin();
Expand Down