Prechádzať zdrojové kódy

HUE-613 [fb] Support snappy compression

Abraham Elmahrek 12 rokov pred
rodič
commit
88550246a6

+ 23 - 13
apps/filebrowser/src/filebrowser/views.py

@@ -688,10 +688,11 @@ def read_contents(codec_type, path, fs, offset, length):
         fhandle = fs.open(path)
         stats = fs.stats(path)
 
-        # Auto codec detection for [gzip, avro, none]
-        # Only done when codec_type is unset
-        contents = fhandle.read(3)
-        if not codec_type:
+        # Auto codec detection for [gzip, avro, snappy, snappy avro, none]
+        if codec_type == 'avro' and snappy_installed() and detect_snappy(fhandle.read()):
+            codec_type = 'snappy_avro'
+        elif not codec_type:
+            contents = fhandle.read(3)
             codec_type = 'none'
             if path.endswith('.gz') and detect_gzip(contents):
                 codec_type = 'gzip'
@@ -699,24 +700,23 @@ def read_contents(codec_type, path, fs, offset, length):
             elif path.endswith('.avro'):
                 if detect_avro(contents):
                     codec_type = 'avro'
-                elif snappy_installed():
-                    if stats.size > MAX_SNAPPY_DECOMPRESSION_SIZE.get():
-                        raise PopupException(_('Failed to validate snappy compressed file. File size is greater than allowed max snappy decompression size of %d.') % MAX_SNAPPY_DECOMPRESSION_SIZE.get())
+                if snappy_installed() and stats.size <= MAX_SNAPPY_DECOMPRESSION_SIZE.get() and detect_snappy(contents + fhandle.read()):
+                    codec_type = 'snappy_avro'
+            elif snappy_installed() and path.endswith('.snappy'):
+                codec_type = 'snappy'
+            elif snappy_installed() and stats.size <= MAX_SNAPPY_DECOMPRESSION_SIZE.get() and detect_snappy(contents + fhandle.read()):
+                codec_type = 'snappy'
 
-                    if detect_snappy(contents + fhandle.read()):
-                        codec_type = 'snappy_avro'
         fhandle.seek(0)
 
-        if codec_type == 'avro' and snappy_installed() and detect_snappy(fhandle.read()):
-            fhandle.seek(0)
-            codec_type = 'snappy_avro'
-
         if codec_type == 'gzip':
             contents = _read_gzip(fhandle, path, offset, length, stats)
         elif codec_type == 'avro':
             contents = _read_avro(fhandle, path, offset, length, stats)
         elif codec_type == 'snappy_avro':
             contents = _read_snappy_avro(fhandle, path, offset, length, stats)
+        elif codec_type == 'snappy':
+            contents = _read_snappy(fhandle, path, offset, length, stats)
         else:
             # for 'none' type.
             contents = _read_simple(fhandle, path, offset, length, stats)
@@ -735,6 +735,16 @@ def _decompress_snappy(compressed_content):
         raise PopupException(_('Failed to decompress snappy compressed file.'), detail=e)
 
 
+def _read_snappy(fhandle, path, offset, length, stats):
+    if not snappy_installed():
+        raise PopupException(_('Failed to decompress snappy compressed file. Snappy is not installed.'))
+
+    if stats.size > MAX_SNAPPY_DECOMPRESSION_SIZE.get():
+        raise PopupException(_('Failed to decompress snappy compressed file. File size is greater than allowed max snappy decompression size of %d.') % MAX_SNAPPY_DECOMPRESSION_SIZE.get())
+
+    return _read_simple(StringIO(_decompress_snappy(fhandle.read())), path, offset, length, stats)
+
+
 def _read_snappy_avro(fhandle, path, offset, length, stats):
     if not snappy_installed():
         raise PopupException(_('Failed to decompress snappy compressed file. Snappy is not installed.'))

+ 62 - 3
apps/filebrowser/src/filebrowser/views_test.py

@@ -592,6 +592,7 @@ def test_view_snappy_compressed_avro():
   import snappy
 
   cluster = pseudo_hdfs4.shared_cluster()
+  finish = []
   try:
     c = make_logged_in_client()
     cluster.fs.setuser(cluster.superuser)
@@ -647,17 +648,75 @@ def test_view_snappy_compressed_avro():
     assert_equal(eval(response.context['view']['contents']), dummy_datum, response)
 
     # Largest snappy compressed file
-    finish = MAX_SNAPPY_DECOMPRESSION_SIZE.set_for_testing(1)
+    finish.append( MAX_SNAPPY_DECOMPRESSION_SIZE.set_for_testing(1) )
     response = c.get('/filebrowser/view/test-snappy-avro-filebrowser/test-view.avro?compression=snappy_avro')
-    assert_true('File size is greater than allowed max snappy decompression size' in response.context['message'], response)
+    assert_true('File size is greater than allowed max snappy decompression size of 1' in response.context['message'], response)
 
   finally:
-    finish()
+    for done in finish:
+      done()
     try:
       cluster.fs.rmtree('/test-snappy-avro-filebrowser/')
     except:
       pass      # Don't let cleanup errors mask earlier failures
 
+
+@attr('requires_hadoop')
+def test_view_snappy_compressed():
+  if not snappy_installed():
+    raise SkipTest
+  import snappy
+
+  cluster = pseudo_hdfs4.shared_cluster()
+  finish = []
+  try:
+    c = make_logged_in_client()
+    cluster.fs.setuser(cluster.superuser)
+    if cluster.fs.isdir('/tmp/test-snappy-filebrowser'):
+      cluster.fs.rmtree('/tmp/test-snappy-filebrowser')
+
+    cluster.fs.mkdir('/tmp/test-snappy-avro-filebrowser/')
+
+    f = cluster.fs.open('/tmp/test-snappy-filebrowser/test-view.snappy', "w")
+    f.write(snappy.compress('This is a test of the emergency broadcasting system.'))
+    f.close()
+
+    f = cluster.fs.open('/tmp/test-snappy-filebrowser/test-view.stillsnappy', "w")
+    f.write(snappy.compress('The broadcasters of your area in voluntary cooperation with the FCC and other authorities.'))
+    f.close()
+
+    f = cluster.fs.open('/tmp/test-snappy-filebrowser/test-view.notsnappy', "w")
+    f.write('foobar')
+    f.close()
+
+    # Snappy compressed fail
+    response = c.get('/filebrowser/view/tmp/test-snappy-filebrowser/test-view.notsnappy?compression=snappy')
+    assert_true('Failed to decompress' in response.context['message'], response)
+
+    # Snappy compressed succeed
+    response = c.get('/filebrowser/view/tmp/test-snappy-filebrowser/test-view.snappy')
+    assert_equal('snappy', response.context['view']['compression'])
+    assert_equal(response.context['view']['contents'], 'This is a test of the emergency broadcasting system.', response)
+
+    # Snappy compressed succeed
+    response = c.get('/filebrowser/view/tmp/test-snappy-filebrowser/test-view.stillsnappy')
+    assert_equal('snappy', response.context['view']['compression'])
+    assert_equal(response.context['view']['contents'], 'The broadcasters of your area in voluntary cooperation with the FCC and other authorities.', response)
+
+    # Largest snappy compressed file
+    finish.append( MAX_SNAPPY_DECOMPRESSION_SIZE.set_for_testing(1) )
+    response = c.get('/filebrowser/view/tmp/test-snappy-filebrowser/test-view.stillsnappy?compression=snappy')
+    assert_true('File size is greater than allowed max snappy decompression size of 1' in response.context['message'], response)
+
+  finally:
+    for done in finish:
+      done()
+    try:
+      cluster.fs.rmtree('/test-snappy-avro-filebrowser/')
+    except:
+      pass      # Don't let cleanup errors mask earlier failures
+
+
 @attr('requires_hadoop')
 def test_view_avro():
   cluster = pseudo_hdfs4.shared_cluster()