Skip to content

Dev #3

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

Merged
xlorne merged 3 commits into from
Nov 12, 2025
Merged

Dev #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
edfa8e5
batch test
xlorne Nov 12, 2025
6c76d04
add #2
xlorne Nov 12, 2025
64af8c8
fix #2
xlorne Nov 12, 2025
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
14 changes: 10 additions & 4 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
<modelVersion>4.0.0</modelVersion>
<groupId>com.codingapi.dbstream</groupId>
<artifactId>dbstream-driver</artifactId>
<version>1.0.8</version>
<version>1.0.9</version>

<url>https://github.com/codingapi/dbstream-driver</url>
<name>dbstream-driver</name>
Expand Down Expand Up @@ -34,8 +34,7 @@
<lombok.version>1.18.42</lombok.version>

<!-- test dependencies properties-->
<springboot.version>2.7.18</springboot.version>
<h2.version>2.2.222</h2.version>
<springboot.version>3.5.7</springboot.version>
</properties>

<dependencies>
Expand All @@ -55,7 +54,14 @@
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<version>${h2.version}</version>
<version>2.2.222</version>
<scope>test</scope>
</dependency>

<dependency>
<groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId>
<version>42.7.3</version>
<scope>test</scope>
</dependency>

Expand Down
6 changes: 3 additions & 3 deletions src/main/java/com/codingapi/dbstream/driver/DBStreamProxyDriver.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,9 @@
package com.codingapi.dbstream.driver;

import com.codingapi.dbstream.interceptor.SQLRunningContext;
import com.codingapi.dbstream.listener.SQLDeleteExecuteListener;
import com.codingapi.dbstream.listener.SQLInsertExecuteListener;
import com.codingapi.dbstream.listener.SQLUpdateExecuteListener;
import com.codingapi.dbstream.listener.stream.SQLDeleteExecuteListener;
import com.codingapi.dbstream.listener.stream.SQLInsertExecuteListener;
import com.codingapi.dbstream.listener.stream.SQLUpdateExecuteListener;
import com.codingapi.dbstream.proxy.ConnectionProxy;
import com.codingapi.dbstream.scanner.DBMetaContext;
import com.codingapi.dbstream.scanner.DBMetaData;
Expand Down
82 changes: 82 additions & 0 deletions src/main/java/com/codingapi/dbstream/interceptor/SQLExecuteParam.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
package com.codingapi.dbstream.interceptor;

import lombok.Getter;
import lombok.Setter;

import java.util.*;

/**
* SQL执行参数信息
*/
public class SQLExecuteParam {


/**
* SQL参数,integer index模式
*/
@Getter
private final Map<Integer,Object> indexParams;
/**
* SQL参数,string key 模型
*/
@Getter
private final Map<String, Object> mapParams;

/**
* 执行的sql
*/
@Getter
@Setter
private String sql;

public SQLExecuteParam() {
this.indexParams = new HashMap<>();
this.mapParams = new HashMap<>();
}


/**
* 更新sql参数
*
* @param key 参数key
* @param value 参数值
*/
public void setParam(String key, Object value) {
mapParams.put(key, value);
}

/**
* 更新sql参数
*
* @param index 参数索引
* @param value 参数值
*/
public void setParam(int index, Object value) {
indexParams.put(index, value);
}

/**
* 清理参数
*/
public void cleanParams(){
this.indexParams.clear();
this.mapParams.clear();
}

/**
* 获取参数列表
* @return List
*/
public List<Object> getListParams(){
List<Object> list = new ArrayList<>();
if (indexParams.isEmpty()) {
return list;
}
List<Integer> keys = new ArrayList<>(indexParams.keySet());
Collections.sort(keys);
for(Integer key: keys){
list.add(indexParams.get(key));
}
return list;
}
}
141 changes: 115 additions & 26 deletions src/main/java/com/codingapi/dbstream/interceptor/SQLExecuteState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,21 +17,25 @@
public class SQLExecuteState {

/**
* SQL参数,integer index模式
* 执行SQL队列
*/
@Getter
private final Map<Integer,Object> indexParams;
private final List<SQLExecuteParam> sqlExecuteParams;

/**
* SQL参数,string key 模型
* 当前执行对象
*/
private SQLExecuteParam currentExecute;

/**
* 模式判断
*/
@Getter
private final Map<String, Object> mapParams;
private boolean batchMode = false;

/**
* 执行的sql
* 当前绑定sql
*/
@Getter
@Setter
private String sql;

/**
Expand Down Expand Up @@ -78,9 +82,56 @@ public SQLExecuteState(String sql, ConnectionProxy connectionProxy, Statement st
this.connectionProxy = connectionProxy;
this.statement = statement;
this.metaData = metaData;
this.sqlExecuteParams = new ArrayList<>();

this.currentExecute = new SQLExecuteParam();
this.currentExecute.setSql(sql);
this.sqlExecuteParams.add(currentExecute);
}

this.indexParams = new HashMap<>();
this.mapParams = new HashMap<>();
public void setSql(String sql){
this.sql = sql;
if(this.currentExecute!=null) {
this.currentExecute.setSql(sql);
}
}

/**
* 添加任务队列
*
* @param sql 执行sql
*/
public void addBatch(String sql) {
batchMode = true;
SQLExecuteParam executeParam = new SQLExecuteParam();
executeParam.setSql(sql);
this.sqlExecuteParams.add(executeParam);
this.currentExecute = executeParam;
}

/**
* 添加任务队列
*
*/
public void addBatch() {
this.addBatch(this.sql);
}

/**
* 清空队列
*/
public void clearBatch() {
this.sqlExecuteParams.clear();
this.currentExecute = null;
}

/**
* 清理参数设置
*/
public void cleanParams(){
if(this.currentExecute!=null) {
this.currentExecute.cleanParams();
}
}

/**
Expand Down Expand Up @@ -114,7 +165,9 @@ public long getExecuteTimestamp() {
* @param value 参数值
*/
public void setParam(String key, Object value) {
mapParams.put(key, value);
if(this.currentExecute!=null) {
currentExecute.setParam(key, value);
}
}

/**
Expand All @@ -124,24 +177,52 @@ public void setParam(String key, Object value) {
* @param value 参数值
*/
public void setParam(int index, Object value) {
indexParams.put(index, value);
if(this.currentExecute!=null) {
currentExecute.setParam(index, value);
}
}

/**
* 获取参数列表
*
* @return List
*/
public List<Object> getListParams(){
List<Object> list = new ArrayList<>();
if (indexParams.isEmpty()) {
return list;
public List<Object> getListParams() {
if(batchMode){
if(this.sqlExecuteParams.isEmpty()){
return new ArrayList<>();
}
int size = this.sqlExecuteParams.size();
return this.sqlExecuteParams.get(size-2).getListParams();
}
List<Integer> keys = new ArrayList<>(indexParams.keySet());
Collections.sort(keys);
for(Integer key: keys){
list.add(indexParams.get(key));
if(this.currentExecute!=null) {
return currentExecute.getListParams();
}
return list;
return new ArrayList<>();
}


/**
* 获取Batch的SQLExecuteState
* @return List
*/
public List<SQLExecuteState> getBatchSQLExecuteStateList(){
if(this.batchMode){
if(this.sqlExecuteParams.isEmpty()){
return new ArrayList<>();
}
int size = this.sqlExecuteParams.size();
List<SQLExecuteState> list = new ArrayList<>();
List<SQLExecuteParam> paramList = this.sqlExecuteParams.subList(0,size-1);
for(SQLExecuteParam executeParam:paramList){
SQLExecuteState executeState = new SQLExecuteState(executeParam.getSql(), connectionProxy,statement,metaData);
executeState.currentExecute = executeParam;
list.add(executeState);
}
return list;

}
return new ArrayList<>();
}


Expand All @@ -167,13 +248,15 @@ public String getTransactionKey() {

/**
* 查询
*
* @param sql sql
* @return 查询结果
* @throws SQLException 查询异常
*/
public List<Map<String, Object>> query(String sql) throws SQLException {
return this.query(sql,new ArrayList<>());
return this.query(sql, new ArrayList<>());
}

/**
* 查询
*
Expand Down Expand Up @@ -216,10 +299,12 @@ public List<Map<String, Object>> getStatementGenerateKeys(DbTable dbTable) {
Map<String, Object> map = new HashMap<>();
ResultSetMetaData resultSetMetaData = rs.getMetaData();
int columnCount = resultSetMetaData.getColumnCount();
List<DbColumn> primaryKeyColumns = dbTable.getPrimaryColumns();
for (int i = 1; i <= columnCount; i++) {
DbColumn dbColumn = primaryKeyColumns.get(i - 1);
map.put(dbColumn.getName(), rs.getObject(i));
String columName = resultSetMetaData.getColumnName(i);
DbColumn dbColumn = dbTable.getColumnByName(columName);
if (dbColumn != null) {
map.put(dbColumn.getName(), rs.getObject(i));
}
}
list.add(map);
}
Expand All @@ -232,6 +317,7 @@ public List<Map<String, Object>> getStatementGenerateKeys(DbTable dbTable) {

/**
* 获取驱动配置信息
*
* @return Properties
*/
public Properties getDriverProperties() {
Expand All @@ -244,6 +330,7 @@ public Properties getDriverProperties() {

/**
* 获取数据库的jdbcUrl
*
* @return jdbcUrl
*/
public String getJdbcUrl() {
Expand All @@ -256,10 +343,11 @@ public String getJdbcUrl() {

/**
* 获取数据库的jdbcKey
*
* @return jdbcKey
*/
public String getJdbcKey(){
if(metaData==null){
public String getJdbcKey() {
if (metaData == null) {
return null;
}
return metaData.getKeyJdbcKey();
Expand All @@ -268,6 +356,7 @@ public String getJdbcKey(){

/**
* 更新数据库的元数据信息
*
* @param tableName 表名
*/
public void updateMetaData(String tableName) throws SQLException {
Expand Down
Loading