Fix hdfs input/output stream closing; fix append to file tests; test hdfs options

ci/etc/hdfs-site.xml

@@ -14,7 +14,7 @@
   limitations under the License. See accompanying LICENSE file.
 -->
 
-<!-- Put site-specific property overrides in this file. -->
+<!-- This is the client configuration for the HDFS integration tests. -->
 
 <configuration>
 	<property>
@@ -38,7 +38,15 @@
 		<value>false</value>
 	</property>
 	<property>
-		<name>dfs.default.replica</name>
+	<name>dfs.default.replica</name>
 		<value>1</value>
+  	</property>
+	<property>
+    <name>dfs.support.append</name>
+    <value>true</value>
+  </property>
+  <property>
+    <name>dfs.client.block.write.replace-datanode-on-failure.enable</name>
+    <value>false</value>
   </property>
 </configuration>

cpp/src/arrow/io/hdfs.cc

@@ -124,9 +124,13 @@ class HdfsReadableFile::HdfsReadableFileImpl : public HdfsAnyFileImpl {
 
   Status Close() {
     if (is_open_) {
+      // is_open_ must be set to false in the beggining, because the destructor
+      // attempts to close the stream again, and if the first close fails, then
+      // the error doesn't get propagated properly and the second close
+      // initiated by the destructor raises a segfault
+      is_open_ = false;
       int ret = driver_->CloseFile(fs_, file_);
       CHECK_FAILURE(ret, "CloseFile");
-      is_open_ = false;
     }
     return Status::OK();
   }
@@ -252,10 +256,14 @@ class HdfsOutputStream::HdfsOutputStreamImpl : public HdfsAnyFileImpl {
 
   Status Close() {
     if (is_open_) {
+      // is_open_ must be set to false in the beggining, because the destructor
+      // attempts to close the stream again, and if the first close fails, then
+      // the error doesn't get propagated properly and the second close
+      // initiated by the destructor raises a segfault
+      is_open_ = false;
       RETURN_NOT_OK(Flush());
       int ret = driver_->CloseFile(fs_, file_);
       CHECK_FAILURE(ret, "CloseFile");
-      is_open_ = false;
     }
     return Status::OK();
   }

python/pyarrow/_hdfs.pyx

@@ -82,7 +82,7 @@ cdef class HdfsOptions:
         if not isinstance(value, tuple) or len(value) != 2:
             raise TypeError('Endpoint must be a tuple of host port pair')
         self.options.connection_config.host = tobytes(value[0])
-        self.options.connection_config.port = tobytes(value[1])
+        self.options.connection_config.port = int(value[1])
 
     @property
     def driver(self):

python/pyarrow/tests/test_fs.py

@@ -34,6 +34,7 @@ def localfs(request, tempdir):
     return dict(
         fs=LocalFileSystem(),
         pathfn=lambda p: (tempdir / p).as_posix(),
+        allow_copy_file=True,
         allow_move_dir=True,
         allow_append_to_file=True,
     )
@@ -44,6 +45,7 @@ def localfs_with_mmap(request, tempdir):
     return dict(
         fs=LocalFileSystem(use_mmap=True),
         pathfn=lambda p: (tempdir / p).as_posix(),
+        allow_copy_file=True,
         allow_move_dir=True,
         allow_append_to_file=True,
     )
@@ -56,6 +58,7 @@ def subtree_localfs(request, tempdir, localfs):
     return dict(
         fs=SubTreeFileSystem(prefix, localfs['fs']),
         pathfn=prefix.__add__,
+        allow_copy_file=True,
         allow_move_dir=True,
         allow_append_to_file=True,
     )
@@ -80,6 +83,7 @@ def s3fs(request, minio_server):
     return dict(
         fs=fs,
         pathfn=bucket.__add__,
+        allow_copy_file=True,
         allow_move_dir=False,
         allow_append_to_file=False,
     )
@@ -91,6 +95,7 @@ def subtree_s3fs(request, s3fs):
     return dict(
         fs=SubTreeFileSystem(prefix, s3fs['fs']),
         pathfn=prefix.__add__,
+        allow_copy_file=True,
         allow_move_dir=False,
         allow_append_to_file=False,
     )
@@ -110,8 +115,9 @@ def hdfs(request, hdfs_server):
     return dict(
         fs=fs,
         pathfn=lambda p: p,
+        allow_copy_file=False,
         allow_move_dir=True,
-        allow_append_to_file=False,
+        allow_append_to_file=True,
     )
 
 
@@ -156,6 +162,11 @@ def allow_move_dir(request, filesystem_config):
     return filesystem_config['allow_move_dir']
 
 
+@pytest.fixture
+def allow_copy_file(request, filesystem_config):
+    return filesystem_config['allow_copy_file']
+
+
 @pytest.fixture
 def allow_append_to_file(request, filesystem_config):
     return filesystem_config['allow_append_to_file']
@@ -271,16 +282,20 @@ def test_delete_dir(fs, pathfn):
         fs.delete_dir(d)
 
 
-def test_copy_file(fs, pathfn):
+def test_copy_file(fs, pathfn, allow_copy_file):
     s = pathfn('test-copy-source-file')
     t = pathfn('test-copy-target-file')
 
     with fs.open_output_stream(s):
         pass
 
-    fs.copy_file(s, t)
-    fs.delete_file(s)
-    fs.delete_file(t)
+    if allow_copy_file:
+        fs.copy_file(s, t)
+        fs.delete_file(s)
+        fs.delete_file(t)
+    else:
+        with pytest.raises(pa.ArrowNotImplementedError):
+            fs.copy_file(s, t)
 
 
 def test_move_directory(fs, pathfn, allow_move_dir):
@@ -411,7 +426,8 @@ def test_open_append_stream(fs, pathfn, compression, buffer_size, compressor,
         s.write(initial)
 
     if allow_append_to_file:
-        with fs.open_append_stream(p, compression, buffer_size) as f:
+        with fs.open_append_stream(p, compression=compression,
+                                   buffer_size=buffer_size) as f:
             f.write(b'\nnewly added')
 
         with fs.open_input_stream(p) as f:
@@ -421,7 +437,8 @@ def test_open_append_stream(fs, pathfn, compression, buffer_size, compressor,
         assert result == b'already existing\nnewly added'
     else:
         with pytest.raises(pa.ArrowNotImplementedError):
-            fs.open_append_stream(p, compression, buffer_size)
+            fs.open_append_stream(p, compression=compression,
+                                  buffer_size=buffer_size)
 
 
 def test_localfs_options():
@@ -476,3 +493,40 @@ def test_s3_options(minio_server):
     )
     assert options.scheme == 'http'
     assert options.endpoint_override == address
+
+
+@pytest.mark.hdfs
+def test_hdfs_options(hdfs_server):
+    from pyarrow.fs import HdfsOptions
+
+    options = HdfsOptions()
+    assert options.endpoint == ('', 0)
+    options.endpoint = ('localhost', 8080)
+    assert options.endpoint == ('localhost', 8080)
+    with pytest.raises(TypeError):
+        options.endpoint = 'localhost:8000'
+
+    assert options.driver == 'hdfs'
+    options.driver = 'hdfs3'
+    assert options.driver == 'hdfs3'
+    with pytest.raises(ValueError):
+        options.driver = 'unknown'
+
+    assert options.replication == 3
+    options.replication = 2
+    assert options.replication == 2
+
+    assert options.user == ''
+    options.user = 'hdfs'
+    assert options.user == 'hdfs'
+
+    assert options.default_block_size == 0
+    options.default_block_size = 128*1024**2
+    assert options.default_block_size == 128*1024**2
+
+    assert options.buffer_size == 0
+    options.buffer_size = 64*1024
+    assert options.buffer_size == 64*1024
+
+    options = HdfsOptions.from_uri('hdfs://localhost:8080/')
+    assert options.endpoint == ('localhost', 8080)