/*
 * Copyright (C) 2014-2016 LinkedIn Corp. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use
 * this file except in compliance with the License. You may obtain a copy of the
 * License at  http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed
 * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
 * CONDITIONS OF ANY KIND, either express or implied.
 */

package gobblin.publisher;

import java.io.IOException;

import org.apache.hadoop.fs.Path;

import gobblin.configuration.State;
import gobblin.configuration.WorkUnitState;


/**
 * Write output to {finaldir}/{currentTimestamp}
 */
public class TimestampDataPublisher extends BaseDataPublisher {

  private final String timestamp;

  public TimestampDataPublisher(State state) throws IOException {
    super(state);
    timestamp = String.valueOf(System.currentTimeMillis());
  }

  @Override
  protected Path getPublisherOutputDir(WorkUnitState workUnitState, int branchId) {
    Path outputDir = super.getPublisherOutputDir(workUnitState, branchId);
    outputDir = new Path(outputDir, timestamp);
    return outputDir;
  }
}

@@ -564,16 +564,37 @@
 eption;%0A
+import java.util.Set;
 %0Aimport 
@@ -620,16 +620,84 @@
 s.Path;%0A
+import gobblin.util.ParallelRunner;%0Aimport gobblin.util.WriterUtils;
 %0Aimport 
@@ -806,24 +806,43 @@
 aldir%7D/%7B
+dbname.tablename%7D/%7B
 current
-T
+t
 imestamp
@@ -1102,72 +1102,251 @@
 %0A%0A  
-@Override%0A  protected Path getPublisherOutputDir(WorkUnitState w
+/**%0A   * Make sure directory exists before running %7B@link BaseDataPublisher#publishData(WorkUnitState, int, boolean, Set)%7D%0A   * so that tables will be moved one at a time rather than all at once%0A   */%0A  @Override%0A  protected void publishData(W
 orkU
@@ -1349,24 +1349,30 @@
 orkUnitState
+ state
 , int branch
@@ -1373,17 +1373,107 @@
 branchId
-)
+, boolean publishSingleTaskData,%0A      Set%3CPath%3E writerOutputPathsMoved) throws IOException
  %7B%0A    P
@@ -1476,17 +1476,26 @@
    Path 
-o
+publisherO
 utputDir
@@ -1501,96 +1501,1095 @@
 r = 
-super.getPublisherOutputDir(workUnitState, branchId);%0A    outputDir = new Path(outputDir
+getPublisherOutputDir(state, branchId);%0A    if (!this.publisherFileSystemByBranches.get(branchId).exists(publisherOutputDir)) %7B%0A      WriterUtils.mkdirsWithRecursivePermission(this.publisherFileSystemByBranches.get(branchId),%0A          publisherOutputDir, this.permissions.get(branchId));%0A    %7D%0A    super.publishData(state, branchId, publishSingleTaskData, writerOutputPathsMoved);%0A  %7D%0A%0A  /**%0A   * Update destination path to put db and table name in format %22dbname.tablename%22 using %7B@link #getDbTableName(String)%7D%0A   * and include timestamp%0A   *%0A   * Input dst format: %7Bfinaldir%7D/%7BschemaName%7D%0A   * Output dst format: %7Bfinaldir%7D/%7Bdbname.tablename%7D/%7Bcurrenttimestamp%7D%0A   */%0A  @Override%0A  protected void movePath(ParallelRunner parallelRunner, State state, Path src, Path dst, int branchId)%0A      throws IOException %7B%0A%0A    String outputDir = dst.toString().substring(0, dst.toString().lastIndexOf('/'));%0A    String schemaName = dst.toString().substring(dst.toString().lastIndexOf('/') + 1, dst.toString().length());%0A    Path newDst = new Path(new Path(outputDir, getDbTableName(schemaName))
 , ti
@@ -2602,28 +2602,662 @@
 p);%0A
+%0A
     
-return outputDir
+if (!this.publisherFileSystemByBranches.get(branchId).exists(newDst)) %7B%0A      WriterUtils.mkdirsWithRecursivePermission(this.publisherFileSystemByBranches.get(branchId),%0A          newDst.getParent(), this.permissions.get(branchId));%0A    %7D%0A%0A    super.movePath(parallelRunner, state, src, newDst, branchId);%0A  %7D%0A%0A  /**%0A   * Translate schema name to %22dbname.tablename%22 to use in path%0A   *%0A   * @param schemaName In format %22dbname_tablename_xxxxx%22%0A   * @return db and table name in format %22dbname.tablename%22%0A   */%0A  private String getDbTableName(String schemaName) %7B%0A    return schemaName.replaceFirst(%22_%22, %22.%22).substring(0, schemaName.lastIndexOf('_'))
 ;%0A  

Update timestamp publisher output path