// Copyright (c) 2012 Cloudera, Inc. All rights reserved.
package com.cloudera.impala.testutil;

import java.io.File;
import java.io.FileWriter;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.hdfs.protocol.LocatedBlocks;

import com.cloudera.impala.catalog.Catalog;
import com.cloudera.impala.catalog.Db;
import com.cloudera.impala.catalog.HdfsPartition;
import com.cloudera.impala.catalog.HdfsPartition.FileDescriptor;
import com.cloudera.impala.catalog.HdfsTable;
import com.cloudera.impala.catalog.Table;

/**
 * Utility to generate an output file with all the block ids for each table
 * currently in the metastore.  Having the block ids allows us to map hdfs
 * files to filesystem files. This is mostly a hack since hdfs does not
 * willingly expose this information.
 */
public class BlockIdGenerator {

  @SuppressWarnings("deprecation")
  public static void main(String[] args)
      throws Exception {

    if (args.length != 1) {
      throw new Exception("Invalid args: BlockIdGenerator <output_file>");
    }

    File output = new File(args[0]);
    FileWriter writer = new FileWriter(output);

    // Load all tables in the catalog
    Catalog catalog = new Catalog(false);
    Db database = catalog.getDb(null);

    Map<String,Table> tables = database.getTables();
    for (String tableName: tables.keySet()) {

      Table table = database.getTable(tableName);
      // Only do this for hdfs tables
      if (table == null || !(table instanceof HdfsTable)) {
        continue;
      }
      HdfsTable hdfsTable = (HdfsTable)table;
      HdfsConfiguration hdfsConfig = new HdfsConfiguration();

      // Write the output as <tablename>: <blockid1> <blockid2> <etc>
      writer.write(tableName + ":");
      for (HdfsPartition partition: hdfsTable.getPartitions()) {
        List<FileDescriptor> fileDescriptors = partition.getFileDescriptors();
        for (FileDescriptor fd : fileDescriptors) {
          String path = fd.getFilePath();
          Path p = new Path(path);

          // Use a deprecated API to get block ids
          DistributedFileSystem dfs = (DistributedFileSystem)p.getFileSystem(hdfsConfig);
          LocatedBlocks locations = dfs.getClient().getNamenode().getBlockLocations(
              p.toUri().getPath(), 0, fd.getFileLength());

          for (LocatedBlock lb : locations.getLocatedBlocks()) {
            long id = lb.getBlock().getBlockId();
            writer.write(" " + id);
          }
        }
      }
      writer.write("\n");
    }
  }

}

@@ -1229,16 +1229,76 @@
 %0A    %7D%0A%0A
+    HdfsConfiguration hdfsConfig = new HdfsConfiguration();%0A
     File
@@ -1349,16 +1349,48 @@
 writer =
+ null;%0A%0A    try %7B%0A      writer =
  new Fil
@@ -1411,16 +1411,18 @@
 );%0A%0A    
+ 
+ 
 // Load 
@@ -1447,16 +1447,18 @@
 catalog%0A
+  
     Cata
@@ -1487,24 +1487,26 @@
 log(false);%0A
+  
     Db datab
@@ -1537,16 +1537,18 @@
 );%0A%0A    
+  
 Map%3CStri
@@ -1584,24 +1584,26 @@
 etTables();%0A
+  
     for (Str
@@ -1643,16 +1643,18 @@
 %0A%0A      
+  
 Table ta
@@ -1695,16 +1695,18 @@
 ;%0A      
+  
 // Only 
@@ -1731,24 +1731,26 @@
 ables%0A      
+ 
+ 
 if (table ==
@@ -1799,16 +1799,18 @@
         
+  
 continue
@@ -1813,26 +1813,30 @@
 inue;%0A      
+ 
+ 
 %7D%0A
+  
       HdfsTa
@@ -1873,71 +1873,11 @@
 le;%0A
+%0A
   
-    HdfsConfiguration hdfsConfig = new HdfsConfiguration();%0A%0A
     
@@ -1944,24 +1944,26 @@
 %3Cetc%3E%0A      
+  
 writer.write
@@ -1977,24 +1977,26 @@
 ame + %22:%22);%0A
+  
       for (H
@@ -2056,16 +2056,18 @@
         
+  
 List%3CFil
@@ -2125,24 +2125,26 @@
 criptors();%0A
+  
         for 
@@ -2193,16 +2193,18 @@
         
+  
 String p
@@ -2237,16 +2237,18 @@
         
+  
 Path p =
@@ -2275,16 +2275,18 @@
         
+ 
+ 
 // Use a
@@ -2328,16 +2328,18 @@
         
+  
 Distribu
@@ -2358,16 +2358,33 @@
 m dfs = 
+%0A                
 (Distrib
@@ -2427,16 +2427,18 @@
 onfig);%0A
+  
         
@@ -2528,16 +2528,18 @@
         
+  
 p.toUri(
@@ -2572,24 +2572,26 @@
 Length());%0A%0A
+  
           fo
@@ -2655,16 +2655,18 @@
         
+ 
+ 
 long id 
@@ -2699,32 +2699,34 @@
 ();%0A            
+  
 writer.write(%22 %22
@@ -2735,34 +2735,38 @@
  id);%0A          
+ 
+ 
 %7D%0A
+  
         %7D%0A      
@@ -2755,24 +2755,26 @@
           %7D%0A
+  
       %7D%0A    
@@ -2771,24 +2771,26 @@
     %7D%0A      
+  
 writer.write
@@ -2793,24 +2793,90 @@
 rite(%22%5Cn%22);%0A
+      %7D%0A    %7D finally %7B%0A      if (writer != null) writer.close();%0A
     %7D%0A  %7D%0A%0A%7D

Fix to close block id output file.