RubyGems - embulk-input-jdbc - Versions diffs - 0.9.2 → 0.9.3 - Mend

embulk-input-jdbc 0.9.2 → 0.9.3

This diff represents the content of publicly available package versions that have been released to one of the supported registries. The information contained in this diff is provided for informational purposes only and reflects changes between package versions as they appear in their respective public registries.

Files changed (7) hide show

checksums.yaml +4 -4
data/README.md +33 -0
data/classpath/embulk-input-jdbc-0.9.3.jar +0 -0
data/src/main/java/org/embulk/input/jdbc/AbstractJdbcInputPlugin.java +46 -5
data/src/main/java/org/embulk/input/jdbc/JdbcInputConnection.java +114 -22
metadata +3 -3
data/classpath/embulk-input-jdbc-0.9.2.jar +0 -0

checksums.yaml CHANGED

@@ -1,7 +1,7 @@
 ---
 SHA1:
-  metadata.gz: a95b2480f16f4ff5e62ce7ecf7fbfc88f6bded02
-  data.tar.gz: 38cd9b33ba3820ec82bc961eaafe8642c6807d6f
+  metadata.gz: 89f25bb2c92757d6e3f6aad8cdd9721d58eb216c
+  data.tar.gz: f1195041a5488157685ba4f435b5e0e67276f107
 SHA512:
-  metadata.gz: e5ffa68e4a72bd90e97829223d7f02925779f524c55073846ea0c5fd083b7291b81a6d3eeca9103aeaeb30a0eda6dcdffb5a48f1518ab383ba32e918b5bfdc93
-  data.tar.gz: e900e943ba5da276aad6066549664fa67eb219b6a78d3a33673fce54c8232841410b18d469e05cdc086f4f17c123a34d395f61831613d98640494df088fa7b03
+  metadata.gz: 66aded100ed9af8b837e9b9d12cf0d43cda252a5f909b6d0d0bbea97143fd3165d02882c5e54aaa3c1f49ab2e4fd804fa7f4bb33c54a915aca282a0be63588ef
+  data.tar.gz: 95bb8bea452106e630740f6ea239a2556c8bee1c36be9fef6318f2be97efc53eefc8844b342b7aba7ea5e659941b63527528ee6392dba9becbab67c1497eafdb

data/README.md CHANGED

@@ -22,6 +22,7 @@ Generic JDBC input plugin for Embulk loads records from a database using a JDBC
 - **options**: extra JDBC properties (hash, default: {})
 - If you write SQL directly,
   - **query**: SQL to run (string)
+  - **use_raw_query_with_incremental**: If true, you can write optimized query using prepared statement. See [Use incremental loading with raw query](#use-incremental-loading-with-raw-query) for more detail (boolean, default: false)
 - If **query** is not set,
   - **table**: destination table name (string, required)
   - **select**: expression of select (e.g. `id, created_at`) (string, default: "*")
@@ -76,6 +77,38 @@ CREATE INDEX embulk_incremental_loading_index ON table (updated_at, id);
 Recommended usage is to leave `incremental_columns` unset and let this plugin automatically finds an auto-increment primary key. Currently, only strings and integers are supported as incremental_columns.
+### Use incremental loading with raw query
+**IMPORTANT**: This is an advanced feature and assume you have an enough knowledge about incremental loading using Embulk and this plugin
+Normally, you can't write your own query for incremental loading.
+`use_raw_query_with_incremental` option allow you to write raw query for incremental loading. It might be well optimized and faster than SQL statement which is automatically generated by plugin.
+Prepared statement starts with `:` is available instead of fixed value.
+`last_record` value is necessary when you use this option.
+Please use prepared statement that is well distinguishable in SQL statement. Using too simple prepared statement like `:a` might cause SQL parse failure.
+In the following example, prepared statement `:foo_id` will be replaced with value "1" which is specified in `last_record`.
+```yaml
+in:
+  type: jdbc
+  query:
+    SELECT
+      foo.id as foo_id, bar.name
+    FROM
+      foo LEFT JOIN bar ON foo.id = bar.id
+    WHERE
+      foo.hoge IS NOT NULL
+      AND foo.id > :foo_id
+    ORDER BY
+      foo.id ASC
+  use_raw_query_with_incremental: true
+  incremental_columns:
+    - foo_id
+  incremental: true
+  last_record: [1]
+```
 ## Example

data/classpath/embulk-input-jdbc-0.9.3.jar ADDED

Binary file

data/src/main/java/org/embulk/input/jdbc/AbstractJdbcInputPlugin.java CHANGED

@@ -6,12 +6,14 @@ import java.net.MalformedURLException;
 import java.net.URISyntaxException;
 import java.net.URL;
 import java.nio.file.Path;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.nio.file.Paths;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.TreeMap;
 import org.slf4j.Logger;
@@ -67,6 +69,10 @@ public abstract class AbstractJdbcInputPlugin
         @ConfigDefault("null")
         public Optional<String> getQuery();
+        @Config("use_raw_query_with_incremental")
+        @ConfigDefault("false")
+        public boolean getUseRawQueryWithIncremental();
         @Config("select")
         @ConfigDefault("null")
         public Optional<String> getSelect();
@@ -210,7 +216,29 @@ public abstract class AbstractJdbcInputPlugin
         // build SELECT query and gets schema of its result
         String rawQuery = getRawQuery(task, con);
-        JdbcSchema querySchema = con.getSchemaOfQuery(rawQuery);
+        JdbcSchema querySchema = null;
+        if (task.getUseRawQueryWithIncremental()) {
+            String temporaryQuery = rawQuery;
+            // Insert pair of columnName:columnIndex order by column name length DESC
+            TreeMap<String, Integer> columnNames = new TreeMap<>(new Comparator<String>() {
+                @Override
+                public int compare(String val1, String val2) {
+                    return val2.length() - val1.length();
+                }
+            });
+            for (int i = 0; i < task.getIncrementalColumns().size(); i++) {
+                columnNames.put(task.getIncrementalColumns().get(i), i);
+            }
+            for (Map.Entry<String, Integer> column : columnNames.entrySet()) {
+                // Temporary replace place holder like ":id" with "?" to avoid SyntaxException while getting schema.
+                temporaryQuery = temporaryQuery.replace(":" + column.getKey(), "?");
+            }
+            querySchema = con.getSchemaOfQuery(temporaryQuery);
+        } else {
+            querySchema = con.getSchemaOfQuery(rawQuery);
+        }
         task.setQuerySchema(querySchema);
         // query schema should not change after incremental query
@@ -251,13 +279,13 @@ public abstract class AbstractJdbcInputPlugin
             }
             if (task.getQuery().isPresent()) {
-                preparedQuery = con.wrapIncrementalQuery(rawQuery, querySchema, incrementalColumnIndexes, lastRecord);
+                preparedQuery = con.wrapIncrementalQuery(rawQuery, querySchema, incrementalColumns, lastRecord, task.getUseRawQueryWithIncremental());
             }
             else {
                 preparedQuery = con.rebuildIncrementalQuery(
                         task.getTable().get(), task.getSelect(),
                         task.getWhere(),
-                        querySchema, incrementalColumnIndexes, lastRecord);
+                        querySchema, incrementalColumns, lastRecord);
             }
         }
         else {
@@ -330,8 +358,21 @@ public abstract class AbstractJdbcInputPlugin
             if (task.getTable().isPresent() || task.getSelect().isPresent() ||
                     task.getWhere().isPresent() || task.getOrderBy().isPresent()) {
                 throw new ConfigException("'table', 'select', 'where' and 'order_by' parameters are unnecessary if 'query' parameter is set.");
-            } else if (!task.getIncrementalColumns().isEmpty() || task.getLastRecord().isPresent()) {
-                throw new ConfigException("'incremental_columns' and 'last_record' parameters are not supported if 'query' parameter is set.");
+            } else if (task.getUseRawQueryWithIncremental()) {
+                String rawQuery = task.getQuery().get();
+                for (String columnName : task.getIncrementalColumns()) {
+                    if (!rawQuery.contains(":" + columnName)) {
+                        throw new ConfigException(String.format("Column \":%s\" doesn't exist in query string", columnName));
+                    }
+                }
+                if (!task.getLastRecord().isPresent()) {
+                    throw new ConfigException("'last_record' is required when 'use_raw_query_with_incremental' is set to true");
+                }
+                if (task.getLastRecord().get().size() != task.getIncrementalColumns().size()) {
+                    throw new ConfigException("size of 'last_record' is different from of 'incremental_columns'");
+                }
+            } else if (!task.getUseRawQueryWithIncremental() && (!task.getIncrementalColumns().isEmpty() || task.getLastRecord().isPresent())) {
+                throw new ConfigException("'incremental_columns' and 'last_record' parameters are not supported if 'query' parameter is set and 'use_raw_query_with_incremental' is set to false.");
             }
             return task.getQuery().get();
         } else if (task.getTable().isPresent()) {

data/src/main/java/org/embulk/input/jdbc/JdbcInputConnection.java CHANGED

@@ -7,17 +7,19 @@ import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.Comparator;
 import java.util.Locale;
+import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Set;
-import org.embulk.config.ConfigException;
 import org.embulk.spi.Exec;
 import org.embulk.input.jdbc.getter.ColumnGetter;
 import org.slf4j.Logger;
 import java.util.List;
 import java.util.ArrayList;
-import static java.util.Locale.ENGLISH;
+import java.util.TreeMap;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -252,7 +254,7 @@ public class JdbcInputConnection
     public PreparedQuery rebuildIncrementalQuery(String tableName,
             Optional<String> selectExpression, Optional<String> whereCondition,
             JdbcSchema querySchema,
-            List<Integer> incrementalColumnIndexes, List<JsonNode> incrementalValues) throws SQLException
+            List<String> incrementalColumns, List<JsonNode> incrementalValues) throws SQLException
     {
         List<JdbcLiteral> parameters = ImmutableList.of();
@@ -268,7 +270,7 @@ public class JdbcInputConnection
             sb.append("(");
             parameters = buildIncrementalConditionTo(sb,
-                    querySchema, incrementalColumnIndexes, incrementalValues);
+                    querySchema, incrementalColumns, incrementalValues);
             sb.append(")");
             newWhereCondition = Optional.of(sb.toString());
@@ -280,7 +282,7 @@ public class JdbcInputConnection
         Optional<String> newOrderByExpression;
         {
             StringBuilder sb = new StringBuilder();
-            buildIncrementalOrderTo(sb, querySchema, incrementalColumnIndexes);
+            buildIncrementalOrderTo(sb, querySchema, incrementalColumns);
             newOrderByExpression = Optional.of(sb.toString());
         }
@@ -292,38 +294,43 @@ public class JdbcInputConnection
     }
     public PreparedQuery wrapIncrementalQuery(String rawQuery, JdbcSchema querySchema,
-            List<Integer> incrementalColumnIndexes, List<JsonNode> incrementalValues) throws SQLException
+            List<String> incrementalColumns, List<JsonNode> incrementalValues,
+                                              boolean useRawQuery) throws SQLException
     {
         StringBuilder sb = new StringBuilder();
         List<JdbcLiteral> parameters = ImmutableList.of();
-        sb.append("SELECT * FROM (");
-        sb.append(truncateStatementDelimiter(rawQuery));
-        sb.append(") embulk_incremental_");
+        if (useRawQuery) {
+            parameters = replacePlaceholder(sb, rawQuery, querySchema, incrementalColumns, incrementalValues);
+        } else {
+            sb.append("SELECT * FROM (");
+            sb.append(truncateStatementDelimiter(rawQuery));
+            sb.append(") embulk_incremental_");
+            if (incrementalValues != null) {
+                sb.append(" WHERE ");
+                parameters = buildIncrementalConditionTo(sb,
+                        querySchema, incrementalColumns, incrementalValues);
+            }
-        if (incrementalValues != null) {
-            sb.append(" WHERE ");
-            parameters = buildIncrementalConditionTo(sb,
-                    querySchema, incrementalColumnIndexes, incrementalValues);
+            sb.append(" ORDER BY ");
+            buildIncrementalOrderTo(sb, querySchema, incrementalColumns);
         }
-        sb.append(" ORDER BY ");
-        buildIncrementalOrderTo(sb, querySchema, incrementalColumnIndexes);
         return new PreparedQuery(sb.toString(), parameters);
     }
     private List<JdbcLiteral> buildIncrementalConditionTo(
             StringBuilder sb,
             JdbcSchema querySchema,
-            List<Integer> incrementalColumnIndexes, List<JsonNode> incrementalValues) throws SQLException
+            List<String> incrementalColumns, List<JsonNode> incrementalValues) throws SQLException
     {
         ImmutableList.Builder<JdbcLiteral> parameters = ImmutableList.builder();
         List<String> leftColumnNames = new ArrayList<>();
         List<JdbcLiteral> rightLiterals = new ArrayList<>();
-        for (int n = 0; n < incrementalColumnIndexes.size(); n++) {
-            int columnIndex = incrementalColumnIndexes.get(n);
+        for (int n = 0; n < incrementalColumns.size(); n++) {
+            int columnIndex = findIncrementalColumnIndex(querySchema, incrementalColumns.get(n));
             JsonNode value = incrementalValues.get(n);
             leftColumnNames.add(querySchema.getColumnName(columnIndex));
             rightLiterals.add(new JdbcLiteral(columnIndex, value));
@@ -351,17 +358,102 @@ public class JdbcInputConnection
         return parameters.build();
     }
+    private int findIncrementalColumnIndex(JdbcSchema schema, String incrementalColumn)
+            throws SQLException
+    {
+        Optional<Integer> index = schema.findColumn(incrementalColumn);
+        // must be present because already checked in AbstractJdbcInputPlugin.findIncrementalColumnIndexes .
+        return index.get().intValue();
+    }
+    private List<JdbcLiteral> replacePlaceholder(StringBuilder sb, String rawQuery, JdbcSchema querySchema,
+                                                 List<String> incrementalColumns, List<JsonNode> incrementalValues)
+            throws SQLException
+    {
+        // Insert pair of columnName:columnIndex order by column name length DESC
+        TreeMap<String, Integer> columnNames = new TreeMap<>(new Comparator<String>() {
+            @Override
+            public int compare(String val1, String val2) {
+                return val2.length() - val1.length();
+            }
+        });
+        ImmutableList.Builder<JdbcLiteral> parameters = ImmutableList.builder();
+        for (String columnName : incrementalColumns) {
+            int columnIndex = findIncrementalColumnIndex(querySchema, columnName);
+            columnNames.put(columnName, columnIndex);
+        }
+        // Add value of each columns
+        for (Map.Entry<Integer, Integer> columnPosition: generateColumnPositionList(rawQuery, columnNames).entrySet()) {
+            int columnIndex = columnPosition.getValue();
+            JsonNode value = incrementalValues.get(columnIndex);
+            parameters.add(new JdbcLiteral(columnIndex, value));
+        }
+        // Replace placeholder ":column1" string with "?"
+        for (Entry<String, Integer> column : columnNames.entrySet()) {
+            String columnName = column.getKey();
+            while (rawQuery.contains(":" + columnName)) {
+                rawQuery = rawQuery.replaceFirst(":" + columnName, "?");
+            }
+        }
+        sb.append(rawQuery);
+        return parameters.build();
+    }
+    /*
+    * This method parse original query that contains placeholder ":column" and store its index position and columnIndex value in Map
+    *
+    * @param query string that contains placeholder like ":column"
+    * @param pair of columnName:columnIndex sorted by column name length desc ["num2", 1]["num", 0]
+    * @return pair of index position where ":column" appears and columnIndex sorted by index position [65,0][105,0][121,1]
+    *
+    * last_record: [1,101]
+    * SELECT * FROM query_load WHERE
+    *   num IS NOT NULL
+    *   AND num > :num
+    *   AND num2 IS NOT NULL
+    *   OR (num = :num AND num2 > :num2)
+    * ORDER BY num ASC, num2 ASC
+    * in above case, return value will be [65,0][105,0][121,1]
+    */
+    private TreeMap<Integer, Integer> generateColumnPositionList(String rawQuery, TreeMap<String, Integer> columnNames)
+    {
+        TreeMap<Integer, Integer> columnPositionList = new TreeMap<>();
+        for (Entry<String, Integer> column : columnNames.entrySet()) {
+            int lastIndex = 0;
+            while (true) {
+                int index = rawQuery.indexOf(":" + column.getKey(), lastIndex);
+                if (index == -1) {
+                    break;
+                }
+                if (!columnPositionList.containsKey(index)) {
+                    columnPositionList.put(index, column.getValue());
+                }
+                lastIndex = index + 2;
+            }
+        }
+        return columnPositionList;
+    }
     private void buildIncrementalOrderTo(StringBuilder sb,
-            JdbcSchema querySchema, List<Integer> incrementalColumnIndexes)
+            JdbcSchema querySchema, List<String> incrementalColumns) throws SQLException
     {
         boolean first = true;
-        for (int i : incrementalColumnIndexes) {
+        for (String incrementalColumn : incrementalColumns) {
             if (first) {
                 first = false;
             } else {
                 sb.append(", ");
             }
-            sb.append(quoteIdentifierString(querySchema.getColumnName(i)));
+            int columnIndex = findIncrementalColumnIndex(querySchema, incrementalColumn);
+            // the following column name is case sensitive,
+            // so should use actual column name got by DatabaseMetaData.
+            sb.append(quoteIdentifierString(querySchema.getColumnName(columnIndex)));
         }
     }

metadata CHANGED

@@ -1,14 +1,14 @@
 --- !ruby/object:Gem::Specification
 name: embulk-input-jdbc
 version: !ruby/object:Gem::Version
-  version: 0.9.2
+  version: 0.9.3
 platform: ruby
 authors:
 - Sadayuki Furuhashi
 autorequire:
 bindir: bin
 cert_chain: []
-date: 2018-07-03 00:00:00.000000000 Z
+date: 2018-08-10 00:00:00.000000000 Z
 dependencies: []
 description: Selects records from a table.
 email:
@@ -19,7 +19,7 @@ extra_rdoc_files: []
 files:
 - README.md
 - build.gradle
-- classpath/embulk-input-jdbc-0.9.2.jar
+- classpath/embulk-input-jdbc-0.9.3.jar
 - lib/embulk/input/jdbc.rb
 - src/main/java/org/embulk/input/JdbcInputPlugin.java
 - src/main/java/org/embulk/input/jdbc/AbstractJdbcInputPlugin.java

data/classpath/embulk-input-jdbc-0.9.2.jar DELETED

Binary file