{"action":"created","comment":{"url":"https://api.github.com/repos/apache/incubator-apex-malhar/pulls/comments/38710537","id":38710537,"diff_hunk":"@@ -0,0 +1,582 @@\n+/**\n+ * Copyright (C) 2015 DataTorrent, Inc.\n+ *\n+ * Licensed under the Apache License, Version 2.0 (the \"License\");\n+ * you may not use this file except in compliance with the License.\n+ * You may obtain a copy of the License at\n+ *\n+ *         http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+package com.datatorrent.lib.db.jdbc;\n+\n+import java.math.BigDecimal;\n+import java.sql.*;\n+import java.util.List;\n+import java.util.Map;\n+\n+import javax.validation.constraints.Min;\n+import javax.validation.constraints.NotNull;\n+\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import com.google.common.base.Preconditions;\n+import com.google.common.collect.Lists;\n+import com.google.common.collect.Maps;\n+\n+import com.datatorrent.api.Context;\n+import com.datatorrent.api.DefaultOutputPort;\n+import com.datatorrent.api.Operator;\n+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;\n+\n+import com.datatorrent.lib.util.FieldInfo;\n+import com.datatorrent.lib.util.PojoUtils;\n+\n+/**\n+ * A concrete input operator that reads from a database through the JDBC API.<br/>\n+ * This operator by default uses the {@link #fieldInfos} and the table name to construct the sql query.<br/>\n+ *\n+ * A user can provide there own query by setting the {@link #query} property which takes precedence.<br/>\n+ *\n+ * For eg. user can set the query property to a complex one : \"select x1, x2 from t1, t2 where t1.x3 = t2.x3 ;\"<br/>\n+ *\n+ * @displayName Jdbc Input Operator\n+ * @category Input\n+ * @tags database, sql, pojo, jdbc\n+ * @since 2.1.0\n+ */\n+public class JdbcPOJOInputOperator extends AbstractJdbcInputOperator<Object> implements Operator.ActivationListener<Context.OperatorContext>\n+{\n+  private static int DEF_BATCH_SIZE = 100;\n+\n+  //table and query definition\n+  protected String tableName;\n+  protected String whereCondition;\n+  protected String groupByClause;\n+  protected String havingCondition;\n+  protected String orderByExpr;\n+  protected String query;\n+  protected boolean limitSupported;\n+\n+  @NotNull\n+  protected List<FieldInfo> fieldInfos;\n+\n+  private long offset;\n+\n+  @Min(1)\n+  private int batchSize;\n+\n+  private final transient List<ActiveFieldInfo> columnFieldSetters;\n+  private transient boolean windowDone;\n+\n+  protected String columnsExpression;\n+  protected List<Integer> columnDataTypes;\n+\n+  private transient PreparedStatement preparedStatement;\n+  protected transient Class<?> pojoClass;\n+\n+  @OutputPortFieldAnnotation(schemaRequired = true)\n+  public final transient DefaultOutputPort<Object> outputPort = new DefaultOutputPort<Object>()\n+  {\n+    @Override\n+    public void setup(Context.PortContext context)\n+    {\n+      pojoClass = context.getValue(Context.PortContext.TUPLE_CLASS);\n+    }\n+  };\n+\n+  public JdbcPOJOInputOperator()\n+  {\n+    super();\n+    batchSize = DEF_BATCH_SIZE;\n+    limitSupported = true;\n+    columnFieldSetters = Lists.newArrayList();\n+  }\n+\n+  @Override\n+  public void setup(Context.OperatorContext context)\n+  {\n+    Preconditions.checkArgument(query != null || tableName != null, \"both query and table name are not set\");\n+    super.setup(context);\n+\n+    try {\n+      //closing the query statement in super class as it is not needed\n+      queryStatement.close();\n+      if (query == null && columnsExpression == null) {\n+        StringBuilder columns = new StringBuilder();\n+        for (int i = 0; i < fieldInfos.size(); i++) {\n+          columns.append(fieldInfos.get(i).getColumnName());\n+          if (i < fieldInfos.size() - 1) {\n+            columns.append(\",\");\n+          }\n+        }\n+        columnsExpression = columns.toString();\n+        LOG.debug(\"select expr {}\", columnsExpression);\n+      }\n+\n+      preparedStatement = store.connection.prepareStatement(queryToRetrieveData());\n+      if (columnDataTypes == null) {\n+        populateColumnDataTypes();\n+      }\n+    } catch (SQLException e) {\n+      throw new RuntimeException(e);\n+    }\n+\n+    for (FieldInfo fi : fieldInfos) {\n+      columnFieldSetters.add(new ActiveFieldInfo(fi));\n+    }\n+  }\n+\n+  protected void populateColumnDataTypes() throws SQLException\n+  {\n+    columnDataTypes = Lists.newArrayList();\n+    preparedStatement.setLong(1, 0);\n+    preparedStatement.setLong(2, 1);\n+    try (ResultSet rs = preparedStatement.executeQuery()) {\n+      Map<String, Integer> nameToType = Maps.newHashMap();\n+      ResultSetMetaData rsMetaData = rs.getMetaData();\n+      LOG.debug(\"resultSet MetaData column count {}\", rsMetaData.getColumnCount());\n+\n+      for (int i = 1; i <= rsMetaData.getColumnCount(); i++) {\n+        int type = rsMetaData.getColumnType(i);\n+        String name = rsMetaData.getColumnName(i);\n+        LOG.debug(\"column name {} type {}\", name, type);\n+        if (query == null) {\n+          columnDataTypes.add(type);\n+        } else {\n+          //when it is a custom query we need to ensure the types are in the same order as field infos\n+          nameToType.put(name, type);\n+        }\n+      }\n+\n+      if (query != null) {\n+        for (FieldInfo fieldInfo : fieldInfos) {\n+          columnDataTypes.add(nameToType.get(fieldInfo.getColumnName()));\n+        }\n+      }\n+    }\n+  }\n+\n+  @Override\n+  public void beginWindow(long l)\n+  {\n+    windowDone = false;\n+  }\n+\n+  @Override\n+  public void emitTuples()\n+  {\n+    if (!windowDone) {\n+      try {\n+        preparedStatement.setLong(1, offset);\n+        preparedStatement.setLong(2, batchSize);\n+\n+        ResultSet resultSet = preparedStatement.executeQuery();\n+        if (resultSet.next()) {","path":"library/src/main/java/com/datatorrent/lib/db/jdbc/JdbcPOJOInputOperator.java","position":181,"original_position":181,"commit_id":"af203f2694ed7dc893f4671695337b7937d27749","original_commit_id":"309c124508880d92b16ea9177025375854a1d243","user":{"login":"ilooner","id":8020165,"avatar_url":"https://avatars.githubusercontent.com/u/8020165?v=3","gravatar_id":"","url":"https://api.github.com/users/ilooner","html_url":"https://github.com/ilooner","followers_url":"https://api.github.com/users/ilooner/followers","following_url":"https://api.github.com/users/ilooner/following{/other_user}","gists_url":"https://api.github.com/users/ilooner/gists{/gist_id}","starred_url":"https://api.github.com/users/ilooner/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ilooner/subscriptions","organizations_url":"https://api.github.com/users/ilooner/orgs","repos_url":"https://api.github.com/users/ilooner/repos","events_url":"https://api.github.com/users/ilooner/events{/privacy}","received_events_url":"https://api.github.com/users/ilooner/received_events","type":"User","site_admin":false},"body":"I was mistaken on this. You were right.","created_at":"2015-09-04T00:01:52Z","updated_at":"2015-09-04T00:01:52Z","html_url":"https://github.com/apache/incubator-apex-malhar/pull/11#discussion_r38710537","pull_request_url":"https://api.github.com/repos/apache/incubator-apex-malhar/pulls/11","_links":{"self":{"href":"https://api.github.com/repos/apache/incubator-apex-malhar/pulls/comments/38710537"},"html":{"href":"https://github.com/apache/incubator-apex-malhar/pull/11#discussion_r38710537"},"pull_request":{"href":"https://api.github.com/repos/apache/incubator-apex-malhar/pulls/11"}}},"pull_request":{"url":"https://api.github.com/repos/apache/incubator-apex-malhar/pulls/11","id":44046944,"html_url":"https://github.com/apache/incubator-apex-malhar/pull/11","diff_url":"https://github.com/apache/incubator-apex-malhar/pull/11.diff","patch_url":"https://github.com/apache/incubator-apex-malhar/pull/11.patch","issue_url":"https://api.github.com/repos/apache/incubator-apex-malhar/issues/11","number":11,"state":"open","locked":false,"title":"Schema integration with JDBC operators","user":{"login":"chandnisingh","id":4690923,"avatar_url":"https://avatars.githubusercontent.com/u/4690923?v=3","gravatar_id":"","url":"https://api.github.com/users/chandnisingh","html_url":"https://github.com/chandnisingh","followers_url":"https://api.github.com/users/chandnisingh/followers","following_url":"https://api.github.com/users/chandnisingh/following{/other_user}","gists_url":"https://api.github.com/users/chandnisingh/gists{/gist_id}","starred_url":"https://api.github.com/users/chandnisingh/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/chandnisingh/subscriptions","organizations_url":"https://api.github.com/users/chandnisingh/orgs","repos_url":"https://api.github.com/users/chandnisingh/repos","events_url":"https://api.github.com/users/chandnisingh/events{/privacy}","received_events_url":"https://api.github.com/users/chandnisingh/received_events","type":"User","site_admin":false},"body":"","created_at":"2015-09-02T21:49:22Z","updated_at":"2015-09-04T00:01:52Z","closed_at":null,"merged_at":null,"merge_commit_sha":"39db67798a70c7590fd48bcd37b728a07bcd7bed","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/incubator-apex-malhar/pulls/11/commits","review_comments_url":"https://api.github.com/repos/apache/incubator-apex-malhar/pulls/11/comments","review_comment_url":"https://api.github.com/repos/apache/incubator-apex-malhar/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/incubator-apex-malhar/issues/11/comments","statuses_url":"https://api.github.com/repos/apache/incubator-apex-malhar/statuses/af203f2694ed7dc893f4671695337b7937d27749","head":{"label":"chandnisingh:jdbc-schemaIntegration","ref":"jdbc-schemaIntegration","sha":"af203f2694ed7dc893f4671695337b7937d27749","user":{"login":"chandnisingh","id":4690923,"avatar_url":"https://avatars.githubusercontent.com/u/4690923?v=3","gravatar_id":"","url":"https://api.github.com/users/chandnisingh","html_url":"https://github.com/chandnisingh","followers_url":"https://api.github.com/users/chandnisingh/followers","following_url":"https://api.github.com/users/chandnisingh/following{/other_user}","gists_url":"https://api.github.com/users/chandnisingh/gists{/gist_id}","starred_url":"https://api.github.com/users/chandnisingh/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/chandnisingh/subscriptions","organizations_url":"https://api.github.com/users/chandnisingh/orgs","repos_url":"https://api.github.com/users/chandnisingh/repos","events_url":"https://api.github.com/users/chandnisingh/events{/privacy}","received_events_url":"https://api.github.com/users/chandnisingh/received_events","type":"User","site_admin":false},"repo":{"id":41395031,"name":"incubator-apex-malhar","full_name":"chandnisingh/incubator-apex-malhar","owner":{"login":"chandnisingh","id":4690923,"avatar_url":"https://avatars.githubusercontent.com/u/4690923?v=3","gravatar_id":"","url":"https://api.github.com/users/chandnisingh","html_url":"https://github.com/chandnisingh","followers_url":"https://api.github.com/users/chandnisingh/followers","following_url":"https://api.github.com/users/chandnisingh/following{/other_user}","gists_url":"https://api.github.com/users/chandnisingh/gists{/gist_id}","starred_url":"https://api.github.com/users/chandnisingh/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/chandnisingh/subscriptions","organizations_url":"https://api.github.com/users/chandnisingh/orgs","repos_url":"https://api.github.com/users/chandnisingh/repos","events_url":"https://api.github.com/users/chandnisingh/events{/privacy}","received_events_url":"https://api.github.com/users/chandnisingh/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/chandnisingh/incubator-apex-malhar","description":"Mirror of Apache Apex malhar (Incubating)","fork":true,"url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar","forks_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/forks","keys_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/keys{/key_id}","collaborators_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/teams","hooks_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/hooks","issue_events_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/issues/events{/number}","events_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/events","assignees_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/assignees{/user}","branches_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/branches{/branch}","tags_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/tags","blobs_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/git/refs{/sha}","trees_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/git/trees{/sha}","statuses_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/statuses/{sha}","languages_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/languages","stargazers_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/stargazers","contributors_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/contributors","subscribers_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/subscribers","subscription_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/subscription","commits_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/commits{/sha}","git_commits_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/git/commits{/sha}","comments_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/comments{/number}","issue_comment_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/issues/comments{/number}","contents_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/contents/{+path}","compare_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/compare/{base}...{head}","merges_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/merges","archive_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/downloads","issues_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/issues{/number}","pulls_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/pulls{/number}","milestones_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/milestones{/number}","notifications_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/labels{/name}","releases_url":"https://api.github.com/repos/chandnisingh/incubator-apex-malhar/releases{/id}","created_at":"2015-08-26T00:27:55Z","updated_at":"2015-08-26T00:28:04Z","pushed_at":"2015-09-03T23:24:07Z","git_url":"git://github.com/chandnisingh/incubator-apex-malhar.git","ssh_url":"git@github.com:chandnisingh/incubator-apex-malhar.git","clone_url":"https://github.com/chandnisingh/incubator-apex-malhar.git","svn_url":"https://github.com/chandnisingh/incubator-apex-malhar","homepage":null,"size":82146,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:devel-3","ref":"devel-3","sha":"cdc8b28affbcbf8d0a6db050895e79bd8a0b8297","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":41348334,"name":"incubator-apex-malhar","full_name":"apache/incubator-apex-malhar","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/incubator-apex-malhar","description":"Mirror of Apache Apex malhar (Incubating)","fork":false,"url":"https://api.github.com/repos/apache/incubator-apex-malhar","forks_url":"https://api.github.com/repos/apache/incubator-apex-malhar/forks","keys_url":"https://api.github.com/repos/apache/incubator-apex-malhar/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/incubator-apex-malhar/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/incubator-apex-malhar/teams","hooks_url":"https://api.github.com/repos/apache/incubator-apex-malhar/hooks","issue_events_url":"https://api.github.com/repos/apache/incubator-apex-malhar/issues/events{/number}","events_url":"https://api.github.com/repos/apache/incubator-apex-malhar/events","assignees_url":"https://api.github.com/repos/apache/incubator-apex-malhar/assignees{/user}","branches_url":"https://api.github.com/repos/apache/incubator-apex-malhar/branches{/branch}","tags_url":"https://api.github.com/repos/apache/incubator-apex-malhar/tags","blobs_url":"https://api.github.com/repos/apache/incubator-apex-malhar/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/incubator-apex-malhar/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/incubator-apex-malhar/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/incubator-apex-malhar/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/incubator-apex-malhar/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/incubator-apex-malhar/languages","stargazers_url":"https://api.github.com/repos/apache/incubator-apex-malhar/stargazers","contributors_url":"https://api.github.com/repos/apache/incubator-apex-malhar/contributors","subscribers_url":"https://api.github.com/repos/apache/incubator-apex-malhar/subscribers","subscription_url":"https://api.github.com/repos/apache/incubator-apex-malhar/subscription","commits_url":"https://api.github.com/repos/apache/incubator-apex-malhar/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/incubator-apex-malhar/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/incubator-apex-malhar/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/incubator-apex-malhar/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/incubator-apex-malhar/contents/{+path}","compare_url":"https://api.github.com/repos/apache/incubator-apex-malhar/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/incubator-apex-malhar/merges","archive_url":"https://api.github.com/repos/apache/incubator-apex-malhar/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/incubator-apex-malhar/downloads","issues_url":"https://api.github.com/repos/apache/incubator-apex-malhar/issues{/number}","pulls_url":"https://api.github.com/repos/apache/incubator-apex-malhar/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/incubator-apex-malhar/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/incubator-apex-malhar/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/incubator-apex-malhar/labels{/name}","releases_url":"https://api.github.com/repos/apache/incubator-apex-malhar/releases{/id}","created_at":"2015-08-25T07:00:05Z","updated_at":"2015-08-26T17:38:55Z","pushed_at":"2015-09-03T23:24:07Z","git_url":"git://github.com/apache/incubator-apex-malhar.git","ssh_url":"git@github.com:apache/incubator-apex-malhar.git","clone_url":"https://github.com/apache/incubator-apex-malhar.git","svn_url":"https://github.com/apache/incubator-apex-malhar","homepage":null,"size":86474,"stargazers_count":1,"watchers_count":1,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":13,"mirror_url":"git://git.apache.org/incubator-apex-malhar.git","open_issues_count":3,"forks":13,"open_issues":3,"watchers":1,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/incubator-apex-malhar/pulls/11"},"html":{"href":"https://github.com/apache/incubator-apex-malhar/pull/11"},"issue":{"href":"https://api.github.com/repos/apache/incubator-apex-malhar/issues/11"},"comments":{"href":"https://api.github.com/repos/apache/incubator-apex-malhar/issues/11/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/incubator-apex-malhar/pulls/11/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/incubator-apex-malhar/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/incubator-apex-malhar/pulls/11/commits"},"statuses":{"href":"https://api.github.com/repos/apache/incubator-apex-malhar/statuses/af203f2694ed7dc893f4671695337b7937d27749"}}}}