You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by gr...@apache.org on 2020/02/18 21:46:29 UTC
[kudu] 03/03: [python] KUDU-1563. Add support for INSERT_IGNORE
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 48709dea9880240b37f2341e3361b1016883658d
Author: Brock Noland <br...@gmail.com>
AuthorDate: Sun Dec 2 00:40:59 2018 +0000
[python] KUDU-1563. Add support for INSERT_IGNORE
Implements Python support for the `INSERT_IGNORE' operation.
Change-Id: I6c45a50d4b87d8f7c4f0f83fbc72932d056d3a79
Reviewed-on: http://gerrit.cloudera.org:8080/4522
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
---
python/kudu/__init__.py | 2 +-
python/kudu/client.pyx | 28 ++++++++++++++++++++++++++++
python/kudu/libkudu_client.pxd | 5 +++++
python/kudu/tests/test_client.py | 11 ++++++++++-
4 files changed, 44 insertions(+), 2 deletions(-)
diff --git a/python/kudu/__init__.py b/python/kudu/__init__.py
index 989eea1..2dc5387 100644
--- a/python/kudu/__init__.py
+++ b/python/kudu/__init__.py
@@ -16,7 +16,7 @@
# under the License.
from kudu.client import (Client, Table, Scanner, Session, # noqa
- Insert, Update, Delete, Predicate,
+ Insert, InsertIgnore, Update, Delete, Predicate,
TimeDelta, KuduError, ScanTokenBuilder,
ScanToken,
LEADER_ONLY,
diff --git a/python/kudu/client.pyx b/python/kudu/client.pyx
index 368d170..7bc8ce5 100644
--- a/python/kudu/client.pyx
+++ b/python/kudu/client.pyx
@@ -843,6 +843,24 @@ cdef class Table:
"""
return Insert(self, record)
+ def new_insert_ignore(self, record=None):
+ """
+ Create a new InsertIgnore operation. Pass the completed InsertIgnore to a Session.
+ If a record is provided, a PartialRow will be initialized with values
+ from the input record. The record can be in the form of a tuple, dict,
+ or list. Dictionary keys can be either column names, indexes, or a
+ mix of both names and indexes.
+
+ Parameters
+ ----------
+ record : tuple/list/dict
+
+ Returns
+ -------
+ insertIgnore : InsertIgnore
+ """
+ return InsertIgnore(self, record)
+
def new_upsert(self, record=None):
"""
Create a new Upsert operation. Pass the completed Upsert to a Session.
@@ -2843,6 +2861,16 @@ cdef class Insert(WriteOperation):
def __dealloc__(self):
del self.op
+cdef class InsertIgnore(WriteOperation):
+ def __cinit__(self, Table table, record=None):
+ self.op = table.ptr().NewInsertIgnore()
+ self.py_row.row = self.op.mutable_row()
+ if record:
+ self.py_row.from_record(record)
+
+ def __dealloc__(self):
+ del self.op
+
cdef class Upsert(WriteOperation):
def __cinit__(self, Table table, record=None):
diff --git a/python/kudu/libkudu_client.pxd b/python/kudu/libkudu_client.pxd
index 2c65a26..e599941 100644
--- a/python/kudu/libkudu_client.pxd
+++ b/python/kudu/libkudu_client.pxd
@@ -458,6 +458,9 @@ cdef extern from "kudu/client/write_op.h" namespace "kudu::client" nogil:
cdef cppclass KuduInsert(KuduWriteOperation):
pass
+ cdef cppclass KuduInsertIgnore(KuduWriteOperation):
+ pass
+
cdef cppclass KuduUpsert(KuduWriteOperation):
pass
@@ -613,6 +616,7 @@ cdef extern from "kudu/client/client.h" namespace "kudu::client" nogil:
int num_replicas()
KuduInsert* NewInsert()
+ KuduInsertIgnore* NewInsertIgnore()
KuduUpsert* NewUpsert()
KuduUpdate* NewUpdate()
KuduDelete* NewDelete()
@@ -648,6 +652,7 @@ cdef extern from "kudu/client/client.h" namespace "kudu::client" nogil:
Status Apply(KuduWriteOperation* write_op)
Status Apply(KuduInsert* write_op)
+ Status Apply(KuduInsertIgnore* write_op)
Status Apply(KuduUpsert* write_op)
Status Apply(KuduUpdate* write_op)
Status Apply(KuduDelete* write_op)
diff --git a/python/kudu/tests/test_client.py b/python/kudu/tests/test_client.py
index f251033..40ab3b9 100644
--- a/python/kudu/tests/test_client.py
+++ b/python/kudu/tests/test_client.py
@@ -163,7 +163,10 @@ class TestClient(KuduTestBase, unittest.TestCase):
table = self.client.table(self.ex_table)
session = self.client.new_session()
for i in range(nrows):
- op = table.new_insert((i, i*2, 'hello_%d' % i))
+ if i % 2 == 0:
+ op = table.new_insert_ignore((i, i*2, 'hello_%d' % i))
+ else:
+ op = table.new_insert((i, i*2, 'hello_%d' % i))
session.apply(op)
# Cannot apply the same insert twice, C++ client does not indicate an
@@ -189,6 +192,11 @@ class TestClient(KuduTestBase, unittest.TestCase):
session.apply(op)
session.flush()
+ # Insert ignore existing row
+ op = table.new_insert_ignore((3, 1, 'hello_1'))
+ session.apply(op)
+
+
scanner = table.scanner().open()
rows = dict((t[0], t) for t in scanner.read_all_tuples())
assert len(rows) == nrows
@@ -196,6 +204,7 @@ class TestClient(KuduTestBase, unittest.TestCase):
datetime.datetime(2016, 10, 30, 10, 12)
.replace(tzinfo=utc))
assert rows[2] == (2, 222, 'upserted', None)
+ assert rows[3] == (3, 6, 'hello_3', None)
# Delete the rows we just wrote
for i in range(nrows):