add the last_path

src/main/java/org/embulk/input/filename/FilenameInputPlugin.java

@@ -97,6 +97,8 @@ public class FilenameInputPlugin
     private static ArrayList<String> tagList;
     
     private static int chunkSize;
+    
+    private static ArrayList<String> last_p = new ArrayList<String>();
 
     @Override
     public ConfigDiff transaction(ConfigSource config,
@@ -164,6 +166,7 @@ public class FilenameInputPlugin
             // End of sort
             
             allFiles.add(files);
+            last_p.add(files.get(0));
             
         }
         
@@ -182,6 +185,8 @@ public class FilenameInputPlugin
             }
             task.setFiles(oneFile);
             taskCount = oneFile.size();
+            last_p = new ArrayList<String>();
+            last_p.add(allFiles.get(0).get(0));
         } else{
             task.setFiles(allFiles);
             taskCount = allFiles.size();
@@ -207,7 +212,10 @@ public class FilenameInputPlugin
             InputPlugin.Control control)
     {
         control.run(taskSource, schema, taskCount);
-        return Exec.newConfigDiff();
+        ConfigDiff diff = Exec.newConfigDiff();
+        diff.set("last_path",last_p);
+        
+        return diff;
     }
 
     @Override
@@ -233,6 +241,7 @@ public class FilenameInputPlugin
             {
                 int nRead;
                 byte[] data = new byte[chunkSize];
+                String filename = new File(file).getCanonicalPath();
                 FileInputStream dataIn  = new  FileInputStream(file);
                 ByteArrayOutputStream buffer = new ByteArrayOutputStream();
                 
@@ -241,12 +250,18 @@ public class FilenameInputPlugin
                     try (PageBuilder pageBuilder = new PageBuilder(Exec.getBufferAllocator(), schema, output))
                     {
                         pageBuilder.setString(0,buffer.toString());//Base64.encodeBase64String(buffer.toByteArray()));
-                        pageBuilder.setString(1, tagList.get(taskIndex) + new File(file).getCanonicalPath() );
+                        pageBuilder.setString(1, tagList.get(taskIndex) + filename );
                         pageBuilder.addRecord();
                         buffer.flush();
                         pageBuilder.finish();
                     }
                 }
+                if (last_p.size() > 1) {
+                    last_p.set(taskIndex,filename);
+                }
+                else {
+                    last_p.set(0,filename);
+                }
             } catch (IOException ex){
                 ex.printStackTrace();
             }