[python] KUDU-3090 Add support for table ownership

Change-Id: I1c1e12a94459e0f04c769d38d96bffb12f337bda
Reviewed-on: http://gerrit.cloudera.org:8080/16217
Tested-by: Kudu Jenkins
Reviewed-by: Grant Henke <[email protected]>
Reviewed-by: Andrew Wong <[email protected]>

python/kudu/client.pyx

@@ -359,7 +359,8 @@ cdef class Client:
         """
         return from_hybridtime(self.cp.GetLatestObservedTimestamp())
 
-    def create_table(self, table_name, Schema schema, partitioning, n_replicas=None):
+    def create_table(self, table_name, Schema schema, partitioning,
+            n_replicas=None, owner=None):
         """
         Creates a new Kudu table from the passed Schema and options.
 
@@ -382,6 +383,8 @@ cdef class Client:
             self._apply_partitioning(c, partitioning, schema)
             if n_replicas:
                 c.num_replicas(n_replicas)
+            if owner:
+                c.set_owner(tobytes(owner))
             s = c.Create()
             check_status(s)
         finally:
@@ -829,6 +832,11 @@ cdef class Table:
         def __get__(self):
             return len(self.schema)
 
+    property owner:
+        """Name of the owner of the table."""
+        def __get__(self):
+            return frombytes(self.ptr().owner())
+
     def rename(self, new_name):
         raise NotImplementedError
 
@@ -3177,6 +3185,21 @@ cdef class TableAlterer:
             _check_convert_range_bound_type(upper_bound_type)
         )
 
+    def set_owner(self, new_owner):
+        """
+        Set the owner of the table.
+
+        Parameters
+        ----------
+        new_owner : string
+
+        Returns
+        -------
+        self : TableAlterer
+        """
+        self._alterer.SetOwner(tobytes(new_owner))
+        return self
+
     def alter(self):
         """
         Alter table. Returns a new table object upon completion of the alter.

python/kudu/libkudu_client.pxd

@@ -592,6 +592,7 @@ cdef extern from "kudu/client/client.h" namespace "kudu::client" nogil:
         KuduTableCreator& split_rows(vector[const KuduPartialRow*]& split_rows)
         KuduTableCreator& num_replicas(int n_replicas)
         KuduTableCreator& wait(c_bool wait)
+        KuduTableCreator& set_owner(const string& owner)
 
         Status Create()
 
@@ -608,6 +609,7 @@ cdef extern from "kudu/client/client.h" namespace "kudu::client" nogil:
                                              KuduPartialRow* upper_bound,
                                              RangePartitionBound lower_bound_type,
                                              RangePartitionBound upper_bound_type)
+        KuduTableAlterer& SetOwner(const string& new_owner)
 
         KuduTableAlterer& wait(c_bool wait)
 
@@ -619,6 +621,7 @@ cdef extern from "kudu/client/client.h" namespace "kudu::client" nogil:
 
         string& name()
         string& id()
+        string& owner()
         KuduSchema& schema()
         int num_replicas()
 

python/kudu/tests/test_client.py

@@ -35,6 +35,7 @@ def test_table_basics(self):
         self.assertEqual(table.name, self.ex_table)
         self.assertEqual(table.num_columns, len(self.schema))
         self.assertIsNotNone(table.id)
+        self.assertIsNotNone(table.owner)
 
     def test_table_column(self):
         table = self.client.table(self.ex_table)
@@ -153,6 +154,22 @@ def test_create_partitioned_table(self):
             except:
                 pass
 
+    def test_create_table_with_different_owner(self):
+        name = 'table_with_different_owner'
+        try:
+            self.client.create_table(
+                    name, self.schema,
+                    partitioning=Partitioning().add_hash_partitions(['key'], 2),
+                    owner='alice')
+
+            self.assertEqual('alice', self.client.table(name).owner)
+
+        finally:
+            try:
+                self.client.delete_table(name)
+            except:
+                pass
+
     def test_insert_nonexistent_field(self):
         table = self.client.table(self.ex_table)
         op = table.new_insert()
@@ -348,6 +365,21 @@ def test_alter_table_rename(self):
         finally:
             self.client.delete_table('alter-newname')
 
+    def test_alter_table_change_owner(self):
+        name = 'alter-owner'
+        try:
+            self.client.create_table(name,
+                                     self.schema,
+                                     self.partitioning)
+            table = self.client.table(name)
+            alterer = self.client.new_table_alterer(table)
+            table = alterer.set_owner('alice').alter()
+            self.assertEqual('alice', self.client.table(name).owner)
+            with self.assertRaises(TypeError):
+                alterer.set_owner(None).alter()
+        finally:
+            self.client.delete_table(name)
+
     def test_alter_column(self):
         try:
             self.client.create_table('alter-column',