[Cherrypick] [PLUGIN-1950] : Log zero records for Table mode and SQL Statement mode and bump up

pom.xml

@@ -20,7 +20,7 @@
 
   <groupId>io.cdap.plugin</groupId>
   <artifactId>multi-table-plugins</artifactId>
-  <version>1.4.1</version>
+  <version>1.4.2-SNAPSHOT</version>
   <packaging>jar</packaging>
 
   <name>Multiple Table Plugins</name>

src/main/java/io/cdap/plugin/format/DBTableRecordReader.java

@@ -24,6 +24,8 @@
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.RecordReader;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.sql.Connection;
@@ -38,6 +40,8 @@
  * Record reader that reads the entire contents of a database table using JDBC.
  */
 public class DBTableRecordReader extends RecordReader<NullWritable, RecordWrapper> {
+  private static final Logger LOG = LoggerFactory.getLogger(DBTableRecordReader.class);
+
   private final DBTableName tableName;
   private final String tableNameField;
   private final MultiTableConf dbConf;
@@ -85,6 +89,10 @@ public boolean nextKeyValue() throws IOException {
         schema = Schema.recordOf(tableName.getTable(), schemaFields);
       }
       if (!results.next()) {
+        if (pos == 0 && DBTableSplit.DEFAULT_CLAUSE.equals(split.getLowerClause())
+          && DBTableSplit.DEFAULT_CLAUSE.equals(split.getUpperClause())) {
+          LOG.info("Source table '{}' has zero records.", tableName.fullTableName());
+        }
         return false;
       }
 

src/main/java/io/cdap/plugin/format/DBTableSplit.java

@@ -26,7 +26,7 @@
  * A split representing data in a database table.
  */
 public class DBTableSplit extends DataDrivenDBInputFormat.DataDrivenDBInputSplit {
-  private static final String DEFAULT_CLAUSE = "1=1";
+  public static final String DEFAULT_CLAUSE = "1=1";
 
   private DBTableName tableName;
 

src/main/java/io/cdap/plugin/format/MultiTableDBInputFormat.java

@@ -196,6 +196,9 @@ private List<InputSplit> getTableSplits(Connection connection, MultiTableDBConfi
                                                                          columnName,
                                                                          conf.getPluginConf().getWhereClause()))) {
       results.next();
+      if (results.getObject(1) == null && results.getObject(2) == null) {
+        return Collections.singletonList(new DBTableSplit(info.getDbTableName()));
+      }
 
       // Based on the type of the results, use a different mechanism
       // for interpolating split points (i.e., numeric splits, text splits,

src/main/java/io/cdap/plugin/format/SQLStatementRecordReader.java

@@ -89,6 +89,9 @@ public boolean nextKeyValue() throws IOException {
         schema = Schema.recordOf(tableName, schemaFields);
       }
       if (!results.next()) {
+        if (pos == 0) {
+          LOG.info("SQL statement '{}' ('{}') has zero records.", split.getId(), split.getSqlStatement());
+        }
         return false;
       }