Log warning when a table has no records to read

When the multi-table source reads from tables that contain zero rows,
the lack of output can be confusing to debug. This adds a WARN-level
log line with the table name in both DBTableRecordReader and
SQLStatementRecordReader so operators can quickly identify empty tables.

Author: psainics

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,7 @@
  * 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 +88,10 @@ public boolean nextKeyValue() throws IOException {
         schema = Schema.recordOf(tableName.getTable(), schemaFields);
       }
       if (!results.next()) {
+        if (pos == 0) {
+          LOG.warn("Split for table '{}' returned no records. Split query: '{}'",
+                   tableName.getTable(), getQuery());
+        }
         return false;
       }
 

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

@@ -19,7 +19,6 @@
 import io.cdap.cdap.api.data.format.StructuredRecord;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.plugin.DriverCleanup;
-import io.cdap.plugin.format.error.collector.ErrorCollectingMultiSQLStatementInputFormat;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.RecordReader;
@@ -89,6 +88,9 @@ public boolean nextKeyValue() throws IOException {
         schema = Schema.recordOf(tableName, schemaFields);
       }
       if (!results.next()) {
+        if (pos == 0) {
+          LOG.warn("SQL statement '{}' returned no records.", split.getSqlStatement());
+        }
         return false;
       }