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.

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

feat: multithreaded bulk import #235

Merged
merged 26 commits into from
Dec 19, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
26 commits
Select commit Hold shift + click to select a range
1f5d327
feat: Add BulkImport APIs and cron
anku255 Mar 20, 2024
cc2ffe1
fix: PR changes
anku255 Mar 21, 2024
38882d7
fix: PR changes
anku255 Mar 29, 2024
6d4da2e
fix: PR changes
anku255 Mar 29, 2024
c39e8e7
fix: PR changes
anku255 Mar 29, 2024
5634523
fix: PR changes
anku255 Apr 2, 2024
3b0df86
fix: PR changes
anku255 Apr 4, 2024
737db91
fix: Update version and changelog
anku255 Apr 9, 2024
1375803
fix: PR changes
anku255 Apr 10, 2024
c6a9329
Merge branch 'feat/bulk-import-base' into feat/bulk-import-1
anku255 Apr 25, 2024
3ecd8ea
fix: PR changes
anku255 Apr 29, 2024
ecb078b
fix: PR changes
anku255 May 29, 2024
4404916
fix: removing restriction of connection pool size for bulk import
tamassoltesz Sep 20, 2024
9c093fb
fix: actually closing the connection
tamassoltesz Sep 27, 2024
e115ebe
fix: add bulk import retry logic for postgres too
tamassoltesz Sep 30, 2024
bc03e1a
chore: merge master to feature branch
tamassoltesz Oct 1, 2024
a6ed05f
fix: fix failing tests
tamassoltesz Oct 2, 2024
b83a780
chore: current state save
tamassoltesz Nov 5, 2024
dfcdf49
chore: merging master to feature branch
tamassoltesz Nov 5, 2024
a5c740b
fix: fixing merge error with changelog
tamassoltesz Nov 5, 2024
57f7d04
feat: bulk inserting the bulk migration data
tamassoltesz Nov 15, 2024
11bd067
fix: fixes and error handling changes
tamassoltesz Nov 22, 2024
ec02abb
fix: fixing tests
tamassoltesz Nov 26, 2024
05edffc
chore: changelog and build version update
tamassoltesz Nov 27, 2024
12768c9
fix: handling app/tenant not found
tamassoltesz Dec 5, 2024
9803649
fix: review fix
tamassoltesz Dec 18, 2024
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
27 changes: 27 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,33 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).

## [Unreleased]

## [7.3.0]

- Adds tables and queries for Bulk Import

### Migration

```sql
"CREATE TABLE IF NOT EXISTS bulk_import_users (
id CHAR(36),
app_id VARCHAR(64) NOT NULL DEFAULT 'public',
primary_user_id VARCHAR(36),
raw_data TEXT NOT NULL,
status VARCHAR(128) DEFAULT 'NEW',
error_msg TEXT,
created_at BIGINT NOT NULL,
updated_at BIGINT NOT NULL,
CONSTRAINT bulk_import_users_pkey PRIMARY KEY(app_id, id),
CONSTRAINT bulk_import_users__app_id_fkey FOREIGN KEY(app_id) REFERENCES apps(app_id) ON DELETE CASCADE
);

CREATE INDEX IF NOT EXISTS bulk_import_users_status_updated_at_index ON bulk_import_users (app_id, status, updated_at);

CREATE INDEX IF NOT EXISTS bulk_import_users_pagination_index1 ON bulk_import_users (app_id, status, created_at DESC,
id DESC);

CREATE INDEX IF NOT EXISTS bulk_import_users_pagination_index2 ON bulk_import_users (app_id, created_at DESC, id DESC);
```
## [7.2.0] - 2024-10-03

- Compatible with plugin interface version 6.3
Expand Down
2 changes: 1 addition & 1 deletion build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@ plugins {
id 'java-library'
}

version = "7.2.0"
version = "7.3.0"

repositories {
mavenCentral()
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,329 @@
/*
* Copyright (c) 2024, VRAI Labs and/or its affiliates. All rights reserved.
*
* This software is licensed under the Apache License, Version 2.0 (the
* "License") as published by the Apache Software Foundation.
*
* 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 io.supertokens.storage.postgresql;

import java.sql.*;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executor;

/**
* BulkImportProxyConnection is a class implementing the Connection interface, serving as a Connection instance in the bulk import user cronjob.
* This cron extensively utilizes existing queries to import users, all of which internally operate within transactions and those query sometimes
* call the commit/rollback method on the connection.
*
* For the purpose of bulkimport cronjob, we aim to employ a single connection for all queries and rollback any operations in case of query failures.
* To achieve this, we use our own proxy Connection instance and override the commit/rollback/close methods to do nothing.
*/

public class BulkImportProxyConnection implements Connection {
private Connection con = null;

public BulkImportProxyConnection(Connection con) {
this.con = con;
}

@Override
public void close() throws SQLException {
//this.con.close();
//we don't want to close here because we are trying to reuse existing code but also using the same connection
//for bulk importing
}

@Override
public void commit() throws SQLException {
//this.con.commit();
}

@Override
public void rollback() throws SQLException {
//this.con.rollback();
}

public void closeForBulkImportProxyStorage() throws SQLException {
this.con.close();
}

public void commitForBulkImportProxyStorage() throws SQLException {
this.con.commit();
}

public void rollbackForBulkImportProxyStorage() throws SQLException {
this.con.rollback();
}

/* Following methods are unchaged */

@Override
public Statement createStatement() throws SQLException {
return this.con.createStatement();
}

@Override
public PreparedStatement prepareStatement(String sql) throws SQLException {
return this.con.prepareStatement(sql);
}

@Override
public CallableStatement prepareCall(String sql) throws SQLException {
return this.con.prepareCall(sql);
}

@Override
public String nativeSQL(String sql) throws SQLException {
return this.con.nativeSQL(sql);
}

@Override
public void setAutoCommit(boolean autoCommit) throws SQLException {
this.con.setAutoCommit(autoCommit);
}

@Override
public boolean getAutoCommit() throws SQLException {
return this.con.getAutoCommit();
}

@Override
public boolean isClosed() throws SQLException {
return this.con.isClosed();
}

@Override
public DatabaseMetaData getMetaData() throws SQLException {
return this.con.getMetaData();
}

@Override
public void setReadOnly(boolean readOnly) throws SQLException {
this.con.setReadOnly(readOnly);
}

@Override
public boolean isReadOnly() throws SQLException {
return this.con.isReadOnly();
}

@Override
public void setCatalog(String catalog) throws SQLException {
this.con.setCatalog(catalog);
}

@Override
public String getCatalog() throws SQLException {
return this.con.getCatalog();
}

@Override
public void setTransactionIsolation(int level) throws SQLException {
this.con.setTransactionIsolation(level);
}

@Override
public int getTransactionIsolation() throws SQLException {
return this.con.getTransactionIsolation();
}

@Override
public SQLWarning getWarnings() throws SQLException {
return this.con.getWarnings();
}

@Override
public void clearWarnings() throws SQLException {
this.con.clearWarnings();
}

@Override
public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
return this.con.createStatement(resultSetType, resultSetConcurrency);
}

@Override
public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency)
throws SQLException {
return this.con.prepareStatement(sql, resultSetType, resultSetConcurrency);
}

@Override
public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
return this.con.prepareCall(sql, resultSetType, resultSetConcurrency);
}

@Override
public Map<String, Class<?>> getTypeMap() throws SQLException {
return this.con.getTypeMap();
}

@Override
public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
this.con.setTypeMap(map);
}

@Override
public void setHoldability(int holdability) throws SQLException {
this.con.setHoldability(holdability);
}

@Override
public int getHoldability() throws SQLException {
return this.con.getHoldability();
}

@Override
public Savepoint setSavepoint() throws SQLException {
return this.con.setSavepoint();
}

@Override
public Savepoint setSavepoint(String name) throws SQLException {
return this.con.setSavepoint(name);
}

@Override
public void rollback(Savepoint savepoint) throws SQLException {
this.con.rollback(savepoint);
}

@Override
public void releaseSavepoint(Savepoint savepoint) throws SQLException {
this.con.releaseSavepoint(savepoint);
}

@Override
public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability)
throws SQLException {
return this.con.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability);
}

@Override
public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency,
int resultSetHoldability) throws SQLException {
return this.con.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
}

@Override
public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency,
int resultSetHoldability) throws SQLException {
return this.con.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
}

@Override
public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException {
return this.con.prepareStatement(sql, autoGeneratedKeys);
}

@Override
public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException {
return this.con.prepareStatement(sql, columnIndexes);
}

@Override
public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException {
return this.con.prepareStatement(sql, columnNames);
}

@Override
public Clob createClob() throws SQLException {
return this.con.createClob();
}

@Override
public Blob createBlob() throws SQLException {
return this.con.createBlob();
}

@Override
public NClob createNClob() throws SQLException {
return this.con.createNClob();
}

@Override
public SQLXML createSQLXML() throws SQLException {
return this.con.createSQLXML();
}

@Override
public boolean isValid(int timeout) throws SQLException {
return this.con.isValid(timeout);
}

@Override
public void setClientInfo(String name, String value) throws SQLClientInfoException {
this.con.setClientInfo(name, value);
}

@Override
public void setClientInfo(Properties properties) throws SQLClientInfoException {
this.con.setClientInfo(properties);
}

@Override
public String getClientInfo(String name) throws SQLException {
return this.con.getClientInfo(name);
}

@Override
public Properties getClientInfo() throws SQLException {
return this.con.getClientInfo();
}

@Override
public Array createArrayOf(String typeName, Object[] elements) throws SQLException {
return this.con.createArrayOf(typeName, elements);
}

@Override
public Struct createStruct(String typeName, Object[] attributes) throws SQLException {
return this.con.createStruct(typeName, attributes);
}

@Override
public void setSchema(String schema) throws SQLException {
this.con.setSchema(schema);
}

@Override
public String getSchema() throws SQLException {
return this.con.getSchema();
}

@Override
public void abort(Executor executor) throws SQLException {
this.con.abort(executor);
}

@Override
public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {
this.con.setNetworkTimeout(executor, milliseconds);
}

@Override
public int getNetworkTimeout() throws SQLException {
return this.con.getNetworkTimeout();
}

@Override
public <T> T unwrap(Class<T> iface) throws SQLException {
return this.con.unwrap(iface);
}

@Override
public boolean isWrapperFor(Class<?> iface) throws SQLException {
return this.con.isWrapperFor(iface);
}
}
Loading
Loading