Răsfoiți Sursa

HUE-8978 [abfs] Fix Chmod, result export & editor autocomplete (#962)

* Changed append to be compatible with exporting to tables
Changed all previous appends to _append to remain compatible with previous code

* Changed code to have more consistent spacing in paramters

* Added a space after every % refering to a '%s'

* Added the ability for ABFS to show up as an autocomplete result

* Fixed issue where Chmod does not work
Slightly changed stats to include group and user

* Fixed issue where the different filesystems such as adl, s3 and abfs did not autocomplete correctly

* Added the ability for Hue to change a csv file to a directory that leads to a csv file

* Added the ability for ABFS to export tables to Hive

* Switched abfspath to a file that does not rely on a class

* Added the ability for the editor to do a full export to ABFS
travisle22 6 ani în urmă
părinte
comite
976e54d788

+ 2 - 0
desktop/core/src/desktop/js/ko/components/assist/assistStorageEntry.js

@@ -145,6 +145,7 @@ class AssistStorageEntry {
         self.entries(
           filteredFiles.map(file => {
             return new AssistStorageEntry({
+              originalType: self.originalType,
               type: self.type,
               definition: file,
               parent: self
@@ -262,6 +263,7 @@ class AssistStorageEntry {
             filteredFiles.map(
               file =>
                 new AssistStorageEntry({
+                  originalType: self.originalType,
                   type: self.type,
                   definition: file,
                   parent: self

+ 1 - 0
desktop/core/src/desktop/js/ko/components/contextPopover/storageContext.js

@@ -49,6 +49,7 @@ class StorageContext {
       do {
         result.unshift({
           name: currentEntry.definition.name,
+          originalType: currentEntry.originalType,
           isActive: currentEntry === self.storageEntry(),
           storageEntry: currentEntry,
           makeActive: function() {

+ 11 - 0
desktop/core/src/desktop/js/sql/autocompleteResults.js

@@ -1308,6 +1308,14 @@ class AutocompleteResults {
             popular: ko.observable(false),
             details: null
           },
+          {
+            value: 'abfs://',
+            meta: META_I18n.keyword,
+            category: CATEGORIES.KEYWORD,
+            weightAdjust: 0,
+            popular: ko.observable(false),
+            details: null
+          },
           {
             value: '/',
             meta: META_I18n.dir,
@@ -1326,6 +1334,9 @@ class AutocompleteResults {
       } else if (/^adl:\/\//i.test(path)) {
         fetchFunction = 'fetchAdlsPath';
         path = path.substring(5);
+      } else if (/^abfs:\/\//i.test(path)) {
+        fetchFunction = 'fetchAbfsPath';
+        path = path.substring(6);
       } else if (/^hdfs:\/\//i.test(path)) {
         path = path.substring(6);
       }

+ 26 - 2
desktop/libs/azure/src/azure/abfs/__init__.py

@@ -25,12 +25,15 @@ import time
 
 from nose.tools import assert_not_equal
 from hadoop.fs import normpath as fs_normpath
+from azure.conf import get_default_abfs_fs
 
 LOG = logging.getLogger(__name__)
 
-ABFS_PATH_RE = re.compile('^/*[aA][bB][fF][sS]{1,2}://([$a-z0-9](?!.*--)[-a-z0-9]{1,61}[a-z0-9])(/(.*?)/?)?$') # bug here
+ABFS_PATH_RE = re.compile('^/*[aA][bB][fF][sS]{1,2}://([$a-z0-9](?!.*--)[-a-z0-9]{1,61}[a-z0-9])(/(.*?)/?)?$')
+ABFS_PATH_FULL = re.compile('^/*[aA][bB][fF][sS]{1,2}://([$a-z0-9](?!.*--)[-a-z0-9]{1,61}[a-z0-9])@[^.]*?\.dfs\.core\.windows\.net(/(.*?)/?)?$')#bug here
 ABFS_ROOT_S = 'abfss://'
 ABFS_ROOT = 'abfs://'
+ABFSACCOUNT_NAME = re.compile('^/*[aA][bB][fF][sS]{1,2}://[$a-z0-9](?!.*--)[-a-z0-9]{1,61}[a-z0-9](@.*?)$')
 
 def parse_uri(uri):
   """
@@ -39,7 +42,9 @@ def parse_uri(uri):
   """
   match = ABFS_PATH_RE.match(uri)
   if not match:
-    raise ValueError("Invalid ABFS URI: %s" % uri)
+    match = ABFS_PATH_FULL.match(uri)
+    if not match:
+      raise ValueError("Invalid ABFS URI: %s" % uri)
   direct_name = match.group(3) or ''
   base_direct_name = match.group(2) or ''
   return match.group(1), direct_name, base_direct_name
@@ -124,6 +129,25 @@ def join(first,*complist):
   return joined
 
 
+def abfspath(path, fs_defaultfs = get_default_abfs_fs()):
+  """
+  Converts a path to a path that the ABFS driver can use
+  """
+  filesystem, dir_name = ("","")
+  try:
+    filesystem, dir_name = parse_uri(path)[:2]
+  except:
+    return path
+  account_name = ABFSACCOUNT_NAME.match(fs_defaultfs)
+  LOG.debug("%s" % fs_defaultfs)
+  if account_name:
+    if path.lower().startswith(ABFS_ROOT):
+      path = ABFS_ROOT + filesystem + account_name.group(1) + '/' + dir_name
+    else:
+      path = ABFS_ROOT_S + filesystem + account_name.group(1) + '/' + dir_name
+  LOG.debug("%s" % path)
+  return path
+
 def abfsdatetime_to_timestamp(datetime):
   """
   Returns timestamp (seconds) by datetime string from ABFS API responses.

+ 65 - 52
desktop/libs/azure/src/azure/abfs/abfs.py

@@ -24,6 +24,8 @@ from builtins import object
 import logging
 import os
 import threading
+import re
+
 from math import ceil
 from posixpath import join
 from urllib.parse import urlparse
@@ -44,7 +46,6 @@ LOG = logging.getLogger(__name__)
 #Azure has a 30MB block limit on upload.
 UPLOAD_CHUCK_SIZE = 30 * 1000 * 1000
 
-
 class ABFSFileSystemException(IOError):
 
   def __init__(self, *args, **kwargs):
@@ -103,6 +104,7 @@ class ABFS(object):
   def _getheaders(self):
     return {
       "Authorization": self._auth_provider.get_token(),
+      "x-ms-version" : "2019-02-02" #note this is required for setaccesscontrols
     }
   
   # Parse info about filesystems, directories, and files
@@ -136,7 +138,7 @@ class ABFS(object):
       raise WebHdfsException
     return True
 
-  def stats(self, path, params = None, **kwargs):
+  def stats(self, path, params=None, **kwargs):
     """
     List the stat of the actual file/directory
     Returns the ABFFStat object
@@ -149,14 +151,14 @@ class ABFS(object):
       return ABFSStat.for_filesystem(self._statsf(file_system, params, **kwargs), path)
     return ABFSStat.for_single(self._stats(file_system + '/' +dir_name, params, **kwargs), path)
   
-  def listdir_stats(self,path, params = None, **kwargs):
+  def listdir_stats(self,path, params=None, **kwargs):
     """
     List the stats for the directories inside the specified path
     Returns the Multiple ABFFStat object #note change later for recursive cases
     """
     if ABFS.isroot(path):
-      LOG.warn("Path: %s is a Filesystem" %path)
-      return self.listfilesystems_stats(params = None, **kwargs)
+      LOG.warn("Path: %s is a Filesystem" % path)
+      return self.listfilesystems_stats(params=None, **kwargs)
     dir_stats = []
     file_system, directory_name = Init_ABFS.parse_uri(path)[:2]
     if params is None:
@@ -166,13 +168,13 @@ class ABFS(object):
     params['resource'] = 'filesystem'
     if directory_name != "":
       params['directory'] = directory_name
-    res = self._root._invoke("GET",file_system, params, headers= self._getheaders(), **kwargs)
+    res = self._root._invoke("GET", file_system, params, headers=self._getheaders(), **kwargs)
     resp = self._root._format_response(res)
     for x in resp['paths']:
       dir_stats.append(ABFSStat.for_directory(res.headers, x, Init_ABFS.ABFS_ROOT +file_system + "/" + x['name']))
     return dir_stats
   
-  def listfilesystems_stats(self, params = None, **kwargs):
+  def listfilesystems_stats(self, params=None, **kwargs):
     """
     Lists the stats inside the File Systems, No functionality for params
     """
@@ -180,13 +182,13 @@ class ABFS(object):
     if params is None:
       params = {}
     params["resource"] = "account"
-    res = self._root._invoke("GET", params = params, headers = self._getheaders() )
+    res = self._root._invoke("GET", params=params, headers=self._getheaders() )
     resp = self._root._format_response(res)
     for x in resp['filesystems']:
       stats.append(ABFSStat.for_filesystems(res.headers, x))
     return stats
   
-  def _stats(self, schemeless_path, params = None, **kwargs):
+  def _stats(self, schemeless_path, params=None, **kwargs):
     """
     Container function for both stats,
     Returns the header of the result
@@ -194,10 +196,11 @@ class ABFS(object):
     if params is None:
       params = {}
     params['action'] = 'getStatus'
-    res = self._root._invoke('HEAD', schemeless_path, params, headers = self._getheaders(), **kwargs)
+    res = self._root._invoke('HEAD', schemeless_path, params, headers=self._getheaders(), **kwargs)
+    #LOG.debug("%s" % res.headers)
     return res.headers
   
-  def _statsf(self, schemeless_path, params = None, **kwargs):
+  def _statsf(self, schemeless_path, params=None, **kwargs):
     """
     Continer function for both stats but if it's a file system
     Returns the header of the result
@@ -205,10 +208,10 @@ class ABFS(object):
     if params is None:
       params = {}
     params['resource'] = 'filesystem'
-    res = self._root._invoke('HEAD', schemeless_path, params, headers = self._getheaders(), **kwargs)
+    res = self._root._invoke('HEAD', schemeless_path, params, headers=self._getheaders(), **kwargs)
     return res.headers
     
-  def listdir(self, path, params = None, glob=None, **kwargs):
+  def listdir(self, path, params=None, glob=None, **kwargs):
     """
     Lists the names inside the current directories 
     """
@@ -219,7 +222,7 @@ class ABFS(object):
     return [x.name for x in listofDir]
   
   
-  def listfilesystems(self, params=None,**kwargs):
+  def listfilesystems(self, params=None, **kwargs):
     """
     Lists the names of the File Systems, limited arguements  
     """
@@ -250,7 +253,7 @@ class ABFS(object):
     """
     return Init_ABFS.normpath(path)
 
-  def open(self, path, option = 'r', *args, **kwargs):
+  def open(self, path, option='r', *args, **kwargs):
     return ABFSFile(self,path, option )
   
   @staticmethod
@@ -269,16 +272,16 @@ class ABFS(object):
 
   # Create Files,directories, or File Systems
   # --------------------------------
-  def mkdir(self, path, params = None, headers = None, *args, **kwargs):
+  def mkdir(self, path, params=None, headers=None, *args, **kwargs):
     """
     Makes a directory
     """
     if params is None:
       params = {}
     params['resource'] = 'directory'
-    self._create_path(path, params = params, headers = params, overwrite = False)
+    self._create_path(path, params=params, headers=params, overwrite=False)
   
-  def create(self, path, overwrite= False, data = None, headers = None, *args, **kwargs):
+  def create(self, path, overwrite=False, data=None, headers=None, *args, **kwargs):
     """
     Makes a File (Put text in data if adding data)
     """
@@ -290,7 +293,7 @@ class ABFS(object):
   def create_home_dir(self, home_path=None):
     raise NotImplementedError("File System not named")
   
-  def _create_path(self,path, params = None, headers = None, overwrite = False):
+  def _create_path(self,path, params=None, headers=None, overwrite=False):
     """
     Container method for Create
     """
@@ -303,29 +306,36 @@ class ABFS(object):
       additional_header.update(headers)
     if not overwrite:
       additional_header['If-None-Match'] = '*'
-    self._root.put(no_scheme,params, headers= additional_header)
+    self._root.put(no_scheme, params, headers=additional_header)
     
   def _create_fs(self, file_system):
     """
     Creates a File System
     """
-    self._root.put(file_system,{'resource': 'filesystem'}, headers= self._getheaders())
+    self._root.put(file_system, {'resource': 'filesystem'}, headers=self._getheaders())
 
   # Read Files
   # --------------------------------
-  def read(self, path, offset = '0', length = 0, *args, **kwargs):
+  def read(self, path, offset='0', length=0, *args, **kwargs):
     """
     Read data from a file
     """
     path = Init_ABFS.strip_scheme(path)
     headers = self._getheaders()
     if length != 0 and length != '0':
-      headers['range']= 'bytes=%s-%s' %(str(offset), str(int(offset) + int(length)))
+      headers['range']= 'bytes=%s-%s' % (str(offset), str(int(offset) + int(length)))
     return self._root.get(path, headers = headers)
   
   # Alter Files
   # --------------------------------
-  def append(self, path, data, size = 0, offset =0 ,params = None, **kwargs):
+  def append(self, path, data, offset=0):
+    if not data:
+      LOG.warn("There is no data to append to")
+      return
+    self._append(path, data)
+    return self.flush(path, {'position' : int(len(data)) + int(offset)})
+  
+  def _append(self, path, data, size=0, offset=0 ,params=None, **kwargs):
     """
     Appends the data to a file
     """
@@ -334,18 +344,20 @@ class ABFS(object):
       LOG.warn("Params not specified, Append will take longer")
       resp = self._stats(path)
       params = {'position' : int(resp['Content-Length']) + offset, 'action' : 'append'}
-      LOG.debug("%s" %params)
+      LOG.debug("%s" % params)
     else:
       params['action'] = 'append'
     headers = {}
-    if size == 0:
+    if size == 0 or size == '0':
       headers['Content-Length'] = str(len(data))
+      if headers['Content-Length'] == '0':
+        return
     else:
       headers['Content-Length'] = str(size)
-    LOG.debug("%s" %headers['Content-Length'])
-    return self._patching_sl( path, params, data, headers,  **kwargs)
+    LOG.debug("%s" % headers)
+    return self._patching_sl( path, params, data, headers, **kwargs)
   
-  def flush(self, path, params = None, headers = None, **kwargs):
+  def flush(self, path, params=None, headers=None, **kwargs):
     """
     Flushes the data(i.e. writes appended data to File)
     """
@@ -360,7 +372,7 @@ class ABFS(object):
     if headers is None:
       headers = {}
     headers['Content-Length'] = '0'
-    self._patching_sl( path, params, header = headers,  **kwargs)
+    self._patching_sl( path, params, header=headers,  **kwargs)
 
   # Remove Filesystems, directories. or Files
   # --------------------------------
@@ -369,15 +381,15 @@ class ABFS(object):
     Removes an item indicated in the path
     Also removes empty directories
     """
-    self._delete(path, recursive = 'false', skip_trash = skip_trash)
+    self._delete(path, recursive='false', skip_trash=skip_trash)
     
-  def rmtree(self, path, skip_trash = True):
+  def rmtree(self, path, skip_trash=True):
     """
     Remove everything in a given directory
     """
-    self._delete(path, recursive = 'true', skip_trash = skip_trash)
+    self._delete(path, recursive='true', skip_trash=skip_trash)
     
-  def _delete(self, path, recursive = 'false', skip_trash=True):
+  def _delete(self, path, recursive='false', skip_trash=True):
     """
     Wrapper function for calling delete, no support for trash or 
     """
@@ -387,19 +399,19 @@ class ABFS(object):
       raise RuntimeError("Cannot Remove Root")
     file_system, dir_name = Init_ABFS.parse_uri(path)[:2]
     if dir_name == '':
-      return self._root.delete(file_system,{'resource': 'filesystem'}, headers= self._getheaders())
+      return self._root.delete(file_system, {'resource': 'filesystem'}, headers=self._getheaders())
     new_path = file_system + '/' + dir_name
     param = None
     if self.isdir(path):
       param = {'recursive' : recursive}
-    self._root.delete(new_path,param , headers= self._getheaders())
+    self._root.delete(new_path, param, headers=self._getheaders())
     
   def restore(self, path):
     raise NotImplementedError("")
   
   # Edit permissions of Filesystems, directories. or Files
   # --------------------------------
-  def chown(self, path, user = None, group = None, *args, **kwargs):
+  def chown(self, path, user=None, group=None, *args, **kwargs):
     """
     Changes ownership (not implemented)
     """
@@ -417,17 +429,17 @@ class ABFS(object):
     header = {}
     if permissionNumber is not None:
       header['x-ms-permissions'] = str(permissionNumber)
-    self.setAccessControl(path, headers = header)
+    self.setAccessControl(path, headers=header)
   
   def setAccessControl(self, path, headers, **kwargs):
     """
     Set Access Controls (Can do both chmod and chown) (not implemented)
     """
     path = Init_ABFS.strip_scheme(path)
-    params= {'action': 'setAccessControl'}
+    params = {'action': 'setAccessControl'}
     if headers is None:
-      headers ={}
-    self._patching_sl( path, params, header = headers,  **kwargs)
+      headers = {}
+    self._patching_sl( path, params, header=headers,  **kwargs)
 
   def mktemp(self, subdir='', prefix='tmp', basedir=None):
     raise NotImplementedError("")
@@ -461,22 +473,22 @@ class ABFS(object):
     Copies the entire contents of a directory to another location
     """
     dst = dst + '/' + Init_ABFS.strip_path(src)
-    LOG.debug("%s" %dst)
+    LOG.debug("%s" % dst)
     self.mkdir(dst)
     other_files = self.listdir(src)
     for x in other_files:
       x = src + '/' + Init_ABFS.strip_path(x)
-      LOG.debug("%s" %x)
+      LOG.debug("%s" % x)
       self.copy(x, dst)
 
   def rename(self, old, new): 
     """
     Renames a file
     """ 
-    LOG.debug("%s\n%s" %(old, new))
+    LOG.debug("%s\n%s" % (old, new))
     headers = {'x-ms-rename-source' : '/' + Init_ABFS.strip_scheme(old) }
     try:
-      self._create_path(new, headers = headers, overwrite= True)
+      self._create_path(new, headers=headers, overwrite=True)
     except WebHdfsException as e:
       if e.code == 409:
         self.copy(old, new)
@@ -524,13 +536,13 @@ class ABFS(object):
       else:
         self._copy_file(local_src, remote_dst)
     
-  def _local_copy_file(self,local_src,remote_dst, chunk_size = UPLOAD_CHUCK_SIZE ):
+  def _local_copy_file(self,local_src,remote_dst, chunk_size=UPLOAD_CHUCK_SIZE ):
     """
     A wraper function for copying local Files
     """
     if os.path.isfile(local_src):
       if self.exists(remote_dst):
-        LOG.info('%s already exists. Skipping.' %remote_dst)
+        LOG.info('%s already exists. Skipping.' % remote_dst)
         return
       else:
         LOG.info('%s does not exist. Trying to copy.' % remote_dst)
@@ -543,7 +555,7 @@ class ABFS(object):
           offset = 0
           while chunk:
             size = len(chunk)
-            self.append(remote_dst, chunk, size = size, params = {'position' : offset})
+            self._append(remote_dst, chunk, size=size, params={'position' : offset})
             offset += size
             chunk = src.read(chunk_size)
           self.flush(remote_dst, params = {'position' : offset})
@@ -612,19 +624,20 @@ class ABFS(object):
         length = chunk_size
       else:
         length = chunk
-      self.append(path, data[i*chunk_size:i*chunk_size + length], length)
-    LOG.debug("%s" %data)
+      self._append(path, data[i*chunk_size:i*chunk_size + length], length)
+    LOG.debug("%s" % data)
     self.flush(path, {'position' : int(size) })
   
   # Use Patch HTTP request
   #----------------------------
-  def _patching_sl(self, schemeless_path, param, data = None, header = None, **kwargs):
+  def _patching_sl(self, schemeless_path, param, data=None, header=None, **kwargs):
     """
     A wraper function for patch
     """
     if header is None:
       header = {}
     header.update(self._getheaders())
-    LOG.debug("%s" %kwargs)
+    LOG.debug("%s" % kwargs)
     return self._root.invoke('PATCH', schemeless_path, param, data, headers = header, **kwargs)
+  
       

+ 32 - 30
desktop/libs/azure/src/azure/abfs/abfs_test.py

@@ -53,36 +53,37 @@ class ABFSTestBase(unittest.TestCase):
     add_to_group('test')
     self.user = User.objects.get(username="test")
       
-    self.test_fs = 'abfs://testfs' + (str(int(time.time()) ))
-    LOG.debug("%s" %self.test_fs)
+    self.test_fs = 'abfs://test' + (str(int(time.time()) ))
+    LOG.debug("%s" % self.test_fs)
     self.client.mkdir(self.test_fs)
 
   def tearDown(self):
     self.client.rmtree(self.test_fs)
     
   def test_list(self):
-    filesystems = self.client.listdir('abfs://')
-    LOG.debug("%s" %filesystems)
+    testfile = 'abfs://'
+    filesystems = self.client.listdir(testfile)
+    LOG.debug("%s" % filesystems)
     assert_true(filesystems is not None, filesystems)
     
-    pathing = self.client.listdir('abfs://' + filesystems[0],  {"recursive" : "true"} )
-    LOG.debug("%s" %pathing)
+    pathing = self.client.listdir(testfile + filesystems[0],  {"recursive" : "true"} )
+    LOG.debug("%s" % pathing)
     assert_true(pathing is not None, pathing)
     
-    directory = self.client.listdir('abfs://' + filesystems[0] + '/' + pathing[0])
-    LOG.debug("%s" %directory)
+    directory = self.client.listdir(testfile + filesystems[0] + '/' + pathing[0])
+    LOG.debug("%s" % directory)
     assert_true(directory is not None, directory)
     
     directory = self.client.listdir(self.test_fs)
-    LOG.debug("%s" %directory)
+    LOG.debug("%s" % directory)
     assert_true(directory is not None, directory)
     
     pathing = self.client._statsf(filesystems[276])
-    LOG.debug("%s" %pathing)
+    LOG.debug("%s" % pathing)
     assert_true(pathing is not None, pathing)
     
     pathing = self.client._statsf(filesystems[277])
-    LOG.debug("%s" %pathing)
+    LOG.debug("%s" % pathing)
     assert_true(pathing is not None, pathing)
     
     
@@ -116,26 +117,26 @@ class ABFSTestBase(unittest.TestCase):
     
     #testing filesystems
     result = self.client.stats(test_fs)
-    LOG.debug("%s" %result)
+    LOG.debug("%s" % result)
     assert_true(result is not None, result)
     result = self.client.listdir_stats(test_fs)
-    LOG.debug("%s" %result)
+    LOG.debug("%s" % result)
     
     #testing directories
     result = self.client.stats(test_dir)
-    LOG.debug("%s" %result)
+    LOG.debug("%s" % result)
     result = self.client.listdir_stats(test_dir)
-    LOG.debug("%s" %result)
+    LOG.debug("%s" % result)
     
     result = self.client.stats(test_dir2)
-    LOG.debug("%s" %result)
+    LOG.debug("%s" % result)
     result = self.client.listdir_stats(test_dir2)
-    LOG.debug("%s" %result)
+    LOG.debug("%s" % result)
     
     result = self.client.stats(test_dir3)
-    LOG.debug("%s" %result)
+    LOG.debug("%s" % result)
     result = self.client.listdir_stats(test_dir3)
-    LOG.debug("%s" %result)
+    LOG.debug("%s" % result)
     
   def test_mkdir(self):
     test_dir = self.test_fs + '/test_mkdir'
@@ -153,10 +154,10 @@ class ABFSTestBase(unittest.TestCase):
     
     test_string = "This is a test."
     test_len = len(test_string)
-    resp = self.client.append(test_file, test_string) #only works with strings
-    LOG.debug("%s" %self.client.stats(test_file))
+    resp = self.client._append(test_file, test_string) #only works with strings
+    LOG.debug("%s" % self.client.stats(test_file))
     try:
-      LOG.debug("%s" %resp)
+      LOG.debug("%s" % resp)
       resp = self.client.read(test_file, length = test_len)
     except:
       LOG.debug("Not written yet")
@@ -200,6 +201,7 @@ class ABFSTestBase(unittest.TestCase):
     self.client.stats(test_file_permission)
     
     self.client.mkdir(test_dir_permission)
+    self.client.chmod(test_dir_permission, '0000')
     self.client.chmod(test_dir_permission, '0777')
     self.client.stats(test_dir_permission)
     
@@ -210,11 +212,11 @@ class ABFSTestBase(unittest.TestCase):
     test_file_permission = test_dir +'/test.txt'
     
     self.client.create(test_file_permission)
-    self.client.chown(test_file_permission, 'temp')
+    self.client.chown(test_file_permission, group = '$superuser' )
     self.client.stats(test_file_permission)
     
     self.client.mkdir(test_dir_permission)
-    self.client.chown(test_dir_permission, 'temp')
+    self.client.chown(test_dir_permission, group = '$superuser')
     self.client.stats(test_dir_permission)
     
   def test_create_with_file_permissions(self):
@@ -261,14 +263,14 @@ class ABFSTestBase(unittest.TestCase):
     
     test_string = "This is a test."
     test_len = len(test_string)
-    resp = self.client.append(test_file, test_string)
+    resp = self.client._append(test_file, test_string)
     self.client.flush(test_file, {"position" : test_len} )
     
     self.client.copy(test_file, testdir2)
     self.client.stats(testdir2 + '/test.txt')
     resp = self.client.read(testdir2 + '/test.txt')
     resp2 = self.client.read(test_file)
-    assert_equal(resp, resp2, "Files %s and %s are not equal" %(test_file, testdir2 + '/test.txt'))
+    assert_equal(resp, resp2, "Files %s and %s are not equal" % (test_file, testdir2 + '/test.txt'))
     
   
   def test_copy_dir(self):
@@ -290,12 +292,12 @@ class ABFSTestBase(unittest.TestCase):
   @staticmethod
   def test_static_methods():
     test_dir = 'abfss://testfs/test_static/'
-    LOG.debug("%s" %test_dir)
+    LOG.debug("%s" % test_dir)
     norm_path = ABFS.normpath(test_dir)
-    LOG.debug("%s" %norm_path)
+    LOG.debug("%s" % norm_path)
     parent = ABFS.parent_path(test_dir)
-    LOG.debug("%s" %parent)
+    LOG.debug("%s" % parent)
     join_path = ABFS.join(test_dir, 'test1')
-    LOG.debug("%s" %join_path)
+    LOG.debug("%s" % join_path)
 
     

+ 1 - 1
desktop/libs/azure/src/azure/abfs/abfsfile.py

@@ -78,7 +78,7 @@ class ABFSFile(object):
     """
     resp = ""
     try:
-      resp = self.fs.read(self.path, offset = self.pos, length = str(length))
+      resp = self.fs.read(self.path, offset=self.pos, length=str(length))
       self.pos += length
     except:
       resp =''

+ 5 - 11
desktop/libs/azure/src/azure/abfs/abfsstats.py

@@ -24,7 +24,7 @@ class ABFSStat(object):
   DIR_MODE = 0o777 | stat.S_IFDIR
   FILE_MODE = 0o666 | stat.S_IFREG
 
-  def __init__(self, isDir, atime, mtime, size, path):
+  def __init__(self, isDir, atime, mtime, size, path, owner = '', group = ''):
     self.name = strip_path(path)
     self.path = path
     self.isDir = isDir
@@ -36,6 +36,8 @@ class ABFSStat(object):
       self.atime = 0
       self.mtime = 0
     self.size = size
+    self.user = owner
+    self.group = group
     
   def __getitem__(self, key):
     try:
@@ -54,14 +56,6 @@ class ABFSStat(object):
   def mode(self):
     return ABFSStat.DIR_MODE if self.isDir else ABFSStat.FILE_MODE
   
-  @property
-  def user(self):
-    return ''
-
-  @property
-  def group(self):
-    return ''
-  
   @property
   def aclBit(self):
     return False
@@ -84,13 +78,13 @@ class ABFSStat(object):
       isDir = resp['isDirectory'] == 'true'
     except:
       isDir = False
-    return cls(isDir, headers['date'], resp['lastModified'], size, path)
+    return cls(isDir, headers['date'], resp['lastModified'], size, path, resp['owner'], resp['group'])
   
   @classmethod
   def for_single(cls,resp, path):
     size = int(resp['Content-Length'])
     isDir = resp['x-ms-resource-type'] == 'directory'
-    return cls(isDir, resp['date'],resp['Last-Modified'], size, path)
+    return cls(isDir, resp['date'],resp['Last-Modified'], size, path, resp['x-ms-owner'], resp['x-ms-group'])
   
   @classmethod
   def for_filesystem(cls, resp, path):

+ 5 - 5
desktop/libs/azure/src/azure/abfs/upload.py

@@ -61,14 +61,14 @@ class ABFSFileUploadHandler(FileUploadHandler):
        # Verify that the path exists
       self._fs.stats(self.destination)
       
-    LOG.debug("Chunk size = %d" %DEFAULT_WRITE_SIZE)
+    LOG.debug("Chunk size = %d" % DEFAULT_WRITE_SIZE)
 
 
   def new_file(self, field_name, file_name, *args, **kwargs):
     if self._is_abfs_upload():
       super(ABFSFileUploadHandler, self).new_file(field_name, file_name, *args, **kwargs)
 
-      LOG.info('Using ABFSFileUploadHandler to handle file upload wit temp file%s.' %file_name)
+      LOG.info('Using ABFSFileUploadHandler to handle file upload wit temp file%s.' % file_name)
       self.target_path = self._fs.join(self.destination, file_name)
       
       try:
@@ -87,8 +87,8 @@ class ABFSFileUploadHandler(FileUploadHandler):
   def receive_data_chunk(self, raw_data, start):
     if self._is_abfs_upload():
       try:
-        LOG.debug("ABFSFileUploadHandler uploading file part with size: %s" %self._part_size)
-        self._fs.append(self.target_path, raw_data, params = {'position' : int(start)})
+        LOG.debug("ABFSFileUploadHandler uploading file part with size: %s" % self._part_size)
+        self._fs._append(self.target_path, raw_data, params = {'position' : int(start)})
         return None
       except Exception as e:
         self._fs.remove(self.target_path)
@@ -103,7 +103,7 @@ class ABFSFileUploadHandler(FileUploadHandler):
       self._fs.flush(self.target_path, {'position' : int(file_size)})
       LOG.info("ABFSFileUploadHandler has completed file upload to ABFS, total file size is: %d." % file_size)
       self.file.size = file_size
-      LOG.debug("%s" %self._fs.stats(self.target_path))
+      LOG.debug("%s" % self._fs.stats(self.target_path))
       return self.file
     else:
       return None

+ 3 - 3
desktop/libs/hadoop/src/hadoop/fs/test_webhdfs.py

@@ -589,7 +589,7 @@ class WebhdfsTests(unittest.TestCase):
     f.close()
     
     resp = self.cluster.fs.listdir(self.prefix)
-    LOG.debug("%s" %resp)
+    LOG.debug("%s" % resp)
     
     test_dir = self.prefix + "/temp2"
     self.cluster.fs.mkdir(test_dir, 0333)
@@ -599,8 +599,8 @@ class WebhdfsTests(unittest.TestCase):
     f.close()
     
     resp = self.cluster.fs.listdir(self.prefix)
-    LOG.debug("%s" %resp)
+    LOG.debug("%s" % resp)
     resp = self.cluster.fs.listdir_stats(self.prefix)
-    LOG.debug("%s" %resp)
+    LOG.debug("%s" % resp)
     self.cluster.fs.remove(test_file)
     self.cluster.fs.remove(test_file2)

+ 5 - 1
desktop/libs/indexer/src/indexer/indexers/sql.py

@@ -27,6 +27,7 @@ from django.utils.translation import ugettext as _
 
 from desktop.lib import django_mako
 from notebook.models import make_notebook
+from azure.abfs.__init__ import abfspath
 
 if sys.version_info[0] > 2:
   from urllib.parse import unquote as urllib_unquote
@@ -142,7 +143,10 @@ class SQLIndexer(object):
           external_path = external_path + '/%s_table' % external_file_name # If dir not just the file, create data dir and move file there.
           self.fs.mkdir(external_path)
           self.fs.rename(source_path, external_path)
-
+    
+    if external_path.lower().startswith("abfs"): #this is to check if its using an ABFS path
+      external_path = abfspath(external_path) 
+      
     sql += django_mako.render_to_string("gen/create_table_statement.mako", {
         'table': {
             'name': table_name,

+ 1 - 1
desktop/libs/indexer/src/indexer/templates/importer.mako

@@ -1734,7 +1734,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
         resizeElements();
       });
       self.isObjectStore = ko.pureComputed(function() {
-        return self.inputFormat() === 'file' && /^(s3a|adl):\/.*$/.test(self.path());
+        return self.inputFormat() === 'file' && /^(s3a|adl|abfs):\/.*$/.test(self.path());
       });
       self.isObjectStore.subscribe(function(newVal) {
         wizard.destination.useDefaultLocation(!newVal);

+ 3 - 0
desktop/libs/notebook/src/notebook/api.py

@@ -45,6 +45,7 @@ from notebook.connectors.oozie_batch import OozieApi
 from notebook.decorators import api_error_handler, check_document_access_permission, check_document_modify_permission
 from notebook.models import escape_rows, make_notebook
 from notebook.views import upgrade_session_properties, get_api
+from azure.abfs.__init__ import abfspath
 
 if sys.version_info[0] > 2:
   import urllib.request, urllib.error
@@ -754,6 +755,8 @@ def export_result(request):
       'allowed': True
     }
   elif data_format == 'hdfs-directory':
+    if destination.lower().startswith("abfs"):
+      destination = abfspath(destination)
     if is_embedded:
       sql, success_url = api.export_large_data_to_hdfs(notebook, snippet, destination)