Merge pull request #134 from redis-developer/master

Merge to main

VERSION

@@ -1 +1 @@
-3.1.5
+3.2.0-SNAPSHOT

build.gradle

@@ -21,7 +21,6 @@ config {
         inceptionYear = '2020'
         vendor = 'Redis'
         tags = ['redis', 'tool', 'import', 'export', 'replication']
-        bytecodeVersion = 8
 
         links {
             website      = "https://github.com/redis-developer/${project.rootProject.name}"
@@ -54,9 +53,7 @@ config {
 
     docs {
         javadoc {
-            autoLinks {
-                enabled = false
-            }
+            enabled = false
         }
     }
 
@@ -98,8 +95,9 @@ subprojects { subproj ->
 
 	    dependencies {
 	        compileOnly group: 'com.google.code.findbugs', name: 'jsr305', version: jsr305Version
-	        testImplementation group: 'commons-io', name: 'commons-io', version: commonsIoVersion
 	        testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
+            testImplementation 'org.junit.jupiter:junit-jupiter-api'
+            testImplementation 'org.junit.jupiter:junit-jupiter-params'
             testImplementation group: 'org.testcontainers', name: 'junit-jupiter', version: testcontainersVersion
 	        testImplementation(group: 'com.redis.testcontainers', name: 'testcontainers-redis', version: testcontainersRedisVersion) {
 	            exclude group: 'com.redis', module: 'lettucemod'

connectors/riot-db/gradle.properties

@@ -0,0 +1,19 @@
+#
+# SPDX-License-Identifier: Apache-2.0
+#
+# Copyright 2022-2023 The RIOT authors.
+#
+# Licensed 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
+#
+#     https://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.
+#
+project_description = RIOT DB
+automatic.module.name = com.redis.riot.db

connectors/riot-db/riot-db.gradle

@@ -0,0 +1,46 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Copyright 2020-2023 The RIOT authors.
+ *
+ * Licensed 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
+ *
+ *     https://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.
+ */
+config {
+    licensing {
+        enabled = false
+    }
+}
+
+dependencies {
+    implementation project(':riot-core')
+    implementation 'org.springframework.boot:spring-boot-autoconfigure'
+    implementation 'org.springframework:spring-jdbc'
+    implementation 'com.mysql:mysql-connector-j'
+    implementation 'org.postgresql:postgresql'
+    implementation group: 'com.microsoft.sqlserver', name: 'mssql-jdbc', version: mssqlVersion
+    implementation group: 'com.oracle.ojdbc', name: 'ojdbc8', version: oracleVersion
+    // implementation group: 'com.ibm.db2', name: 'jcc', version: db2Version
+    // implementation group: 'org.xerial', name: 'sqlite-jdbc', version: sqliteVersion
+}
+
+compileJava {
+    options.compilerArgs += ["-AprojectPath=${project.group}/${project.name}"]
+}
+
+if (!(project.findProperty('automatic.module.name.skip') ?: false).toBoolean()) {
+    jar {
+        manifest {
+            attributes('Automatic-Module-Name': project.findProperty('automatic.module.name'))
+        }
+    }
+}

connectors/riot-db/src/main/java/com/redis/riot/db/DataSourceOptions.java

@@ -0,0 +1,45 @@
+package com.redis.riot.db;
+
+public class DataSourceOptions {
+
+    private String driver;
+
+    private String url;
+
+    private String username;
+
+    private String password;
+
+    public String getDriver() {
+        return driver;
+    }
+
+    public void setDriver(String driver) {
+        this.driver = driver;
+    }
+
+    public String getUrl() {
+        return url;
+    }
+
+    public void setUrl(String url) {
+        this.url = url;
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+}

connectors/riot-db/src/main/java/com/redis/riot/db/DatabaseExport.java

@@ -0,0 +1,85 @@
+package com.redis.riot.db;
+
+import java.util.Map;
+
+import javax.sql.DataSource;
+
+import org.springframework.batch.item.database.JdbcBatchItemWriter;
+import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
+import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
+import org.springframework.lang.Nullable;
+
+import com.redis.riot.core.AbstractMapExport;
+import com.redis.spring.batch.ValueType;
+
+public class DatabaseExport extends AbstractMapExport {
+
+    public static final boolean DEFAULT_ASSERT_UPDATES = true;
+
+    private String sql;
+
+    private DataSourceOptions dataSourceOptions = new DataSourceOptions();
+
+    private boolean assertUpdates = DEFAULT_ASSERT_UPDATES;
+
+    public String getSql() {
+        return sql;
+    }
+
+    public void setSql(String sql) {
+        this.sql = sql;
+    }
+
+    public DataSourceOptions getDataSourceOptions() {
+        return dataSourceOptions;
+    }
+
+    public void setDataSourceOptions(DataSourceOptions dataSourceOptions) {
+        this.dataSourceOptions = dataSourceOptions;
+    }
+
+    public boolean isAssertUpdates() {
+        return assertUpdates;
+    }
+
+    public void setAssertUpdates(boolean assertUpdates) {
+        this.assertUpdates = assertUpdates;
+    }
+
+    @Override
+    protected JdbcBatchItemWriter<Map<String, Object>> writer() {
+        JdbcBatchItemWriterBuilder<Map<String, Object>> writer = new JdbcBatchItemWriterBuilder<>();
+        writer.itemSqlParameterSourceProvider(NullableSqlParameterSource::new);
+        writer.dataSource(dataSource());
+        writer.sql(sql);
+        writer.assertUpdates(assertUpdates);
+        return writer.build();
+    }
+
+    private DataSource dataSource() {
+        return DatabaseUtils.dataSource(dataSourceOptions);
+    }
+
+    private static class NullableSqlParameterSource extends MapSqlParameterSource {
+
+        public NullableSqlParameterSource(@Nullable Map<String, ?> values) {
+            super(values);
+        }
+
+        @Override
+        @Nullable
+        public Object getValue(String paramName) {
+            if (!hasValue(paramName)) {
+                return null;
+            }
+            return super.getValue(paramName);
+        }
+
+    }
+
+    @Override
+    protected ValueType getValueType() {
+        return ValueType.STRUCT;
+    }
+
+}

connectors/riot-db/src/main/java/com/redis/riot/db/DatabaseImport.java

@@ -0,0 +1,133 @@
+package com.redis.riot.db;
+
+import java.util.Map;
+
+import javax.sql.DataSource;
+
+import org.springframework.batch.core.Job;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.database.AbstractCursorItemReader;
+import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
+import org.springframework.jdbc.core.ColumnMapRowMapper;
+
+import com.redis.riot.core.AbstractMapImport;
+import com.redis.riot.core.RiotExecutionContext;
+import com.redis.riot.core.StepBuilder;
+
+public class DatabaseImport extends AbstractMapImport {
+
+    public static final int DEFAULT_FETCH_SIZE = AbstractCursorItemReader.VALUE_NOT_SET;
+
+    public static final int DEFAULT_MAX_RESULT_SET_ROWS = AbstractCursorItemReader.VALUE_NOT_SET;
+
+    public static final int DEFAULT_QUERY_TIMEOUT = AbstractCursorItemReader.VALUE_NOT_SET;
+
+    private String sql;
+
+    private DataSourceOptions dataSourceOptions = new DataSourceOptions();
+
+    private int maxItemCount;
+
+    private int fetchSize = DEFAULT_FETCH_SIZE;
+
+    private int maxResultSetRows = DEFAULT_MAX_RESULT_SET_ROWS;
+
+    private int queryTimeout = DEFAULT_QUERY_TIMEOUT;
+
+    private boolean useSharedExtendedConnection;
+
+    private boolean verifyCursorPosition;
+
+    public String getSql() {
+        return sql;
+    }
+
+    public void setSql(String sql) {
+        this.sql = sql;
+    }
+
+    public DataSourceOptions getDataSourceOptions() {
+        return dataSourceOptions;
+    }
+
+    public void setDataSourceOptions(DataSourceOptions dataSourceOptions) {
+        this.dataSourceOptions = dataSourceOptions;
+    }
+
+    public int getMaxItemCount() {
+        return maxItemCount;
+    }
+
+    public int getFetchSize() {
+        return fetchSize;
+    }
+
+    public int getMaxResultSetRows() {
+        return maxResultSetRows;
+    }
+
+    public int getQueryTimeout() {
+        return queryTimeout;
+    }
+
+    public boolean isUseSharedExtendedConnection() {
+        return useSharedExtendedConnection;
+    }
+
+    public boolean isVerifyCursorPosition() {
+        return verifyCursorPosition;
+    }
+
+    public void setMaxItemCount(int maxItemCount) {
+        this.maxItemCount = maxItemCount;
+    }
+
+    public void setFetchSize(int fetchSize) {
+        this.fetchSize = fetchSize;
+    }
+
+    public void setMaxResultSetRows(int rows) {
+        this.maxResultSetRows = rows;
+    }
+
+    public void setQueryTimeout(int queryTimeout) {
+        this.queryTimeout = queryTimeout;
+    }
+
+    public void setUseSharedExtendedConnection(boolean useSharedExtendedConnection) {
+        this.useSharedExtendedConnection = useSharedExtendedConnection;
+    }
+
+    public void setVerifyCursorPosition(boolean verifyCursorPosition) {
+        this.verifyCursorPosition = verifyCursorPosition;
+    }
+
+    @Override
+    protected Job job(RiotExecutionContext executionContext) {
+        StepBuilder<Map<String, Object>, Map<String, Object>> step = createStep();
+        step.name(getName());
+        step.reader(reader());
+        step.writer(writer(executionContext));
+        return jobBuilder().start(step.build()).build();
+    }
+
+    private ItemReader<Map<String, Object>> reader() {
+        DataSource dataSource = DatabaseUtils.dataSource(dataSourceOptions);
+        JdbcCursorItemReaderBuilder<Map<String, Object>> builder = new JdbcCursorItemReaderBuilder<>();
+        builder.saveState(false);
+        builder.dataSource(dataSource);
+        builder.name("database-reader");
+        builder.rowMapper(new ColumnMapRowMapper());
+        builder.sql(sql);
+        builder.fetchSize(fetchSize);
+        builder.maxRows(maxResultSetRows);
+        builder.queryTimeout(queryTimeout);
+        builder.useSharedExtendedConnection(useSharedExtendedConnection);
+        builder.verifyCursorPosition(verifyCursorPosition);
+        if (maxItemCount > 0) {
+            builder.maxItemCount(maxItemCount);
+        }
+        return builder.build();
+    }
+
+}

connectors/riot-db/src/main/java/com/redis/riot/db/DatabaseUtils.java

@@ -0,0 +1,21 @@
+package com.redis.riot.db;
+
+import javax.sql.DataSource;
+
+import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
+
+public abstract class DatabaseUtils {
+
+    private DatabaseUtils() {
+    }
+
+    static DataSource dataSource(DataSourceOptions options) {
+        DataSourceProperties properties = new DataSourceProperties();
+        properties.setUrl(options.getUrl());
+        properties.setDriverClassName(options.getDriver());
+        properties.setUsername(options.getUsername());
+        properties.setPassword(options.getPassword());
+        return properties.initializeDataSourceBuilder().build();
+    }
+
+}