You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by la...@apache.org on 2023/06/17 16:56:08 UTC
[kudu] branch master updated (8e8a39741 -> 570936fa8)
This is an automated email from the ASF dual-hosted git repository.
laiyingchun pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
from 8e8a39741 KUDU-3413 add tenant info in metadata for multi-tenancy
new 6d48cd6f9 KUDU-1945 Add Python example for non-unique PK
new fdd373ab1 [KUDU-3483] Log requestor info and request id for write interface
new 570936fa8 [Client] Add query id to trace the whole query process
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../basic-python-example/non_unique_primary_key.py | 221 +++++++++++++++++++++
.../kudu/client/AbstractKuduScannerBuilder.java | 16 ++
.../org/apache/kudu/client/AsyncKuduScanner.java | 27 ++-
.../java/org/apache/kudu/client/KuduScanToken.java | 4 +
.../java/org/apache/kudu/client/KuduScanner.java | 2 +-
.../org/apache/kudu/client/TestKuduScanner.java | 39 ++++
.../java/org/apache/kudu/client/TestScanToken.java | 53 +++++
src/kudu/tserver/tablet_service.cc | 5 +-
8 files changed, 364 insertions(+), 3 deletions(-)
create mode 100644 examples/python/basic-python-example/non_unique_primary_key.py
[kudu] 02/03: [KUDU-3483] Log requestor info and request id for write interface
Posted by la...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
laiyingchun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit fdd373ab1be38f4eeb65360661ec039c5ab4a572
Author: xinghuayu007 <14...@qq.com>
AuthorDate: Tue May 23 11:15:16 2023 +0800
[KUDU-3483] Log requestor info and request id for write interface
Currently, it is hard to trace the whole process of inserting data.
Write interface logs the data, but doesn't log the requestor information
and client information.
This patch logs the requestor information about downstream IP and
client information about client id, sequence number for debuging.
Change-Id: I0a3bacd947146fb52f94ab09dda94c96ed6d8ab8
Reviewed-on: http://gerrit.cloudera.org:8080/19950
Tested-by: Kudu Jenkins
Reviewed-by: Yingchun Lai <la...@apache.org>
---
src/kudu/tserver/tablet_service.cc | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/src/kudu/tserver/tablet_service.cc b/src/kudu/tserver/tablet_service.cc
index bed57de7e..a7f50f6cb 100644
--- a/src/kudu/tserver/tablet_service.cc
+++ b/src/kudu/tserver/tablet_service.cc
@@ -1539,7 +1539,10 @@ void TabletServiceImpl::Write(const WriteRequestPB* req,
const auto& tablet_id = req->tablet_id();
TRACE_EVENT1("tserver", "TabletServiceImpl::Write",
"tablet_id", tablet_id);
- DVLOG(3) << "Received Write RPC: " << SecureDebugString(*req);
+ DVLOG(3) << Substitute("Received Write RPC: $0, requestor: $1, request id: $2",
+ SecureDebugString(*req), context->requestor_string(),
+ context->request_id() == nullptr ?
+ "" : SecureDebugString(*context->request_id()));
scoped_refptr<TabletReplica> replica;
if (!LookupRunningTabletReplicaOrRespond(
server_->tablet_manager(), tablet_id, resp, context, &replica)) {
[kudu] 01/03: KUDU-1945 Add Python example for non-unique PK
Posted by la...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
laiyingchun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 6d48cd6f90e7820fbabd0560139ad48fe9d66a29
Author: Marton Greber <gr...@gmail.com>
AuthorDate: Wed May 10 13:56:02 2023 +0000
KUDU-1945 Add Python example for non-unique PK
In commit 3019848d00ac721b30d98a1aeb77bc205352b7b5 a C++ client example
has been added to showcase the usage of the non-unique PK. This patch
translates this particular example to Python.
The C++ examples are tested during upstream submission, however the Java
and Python client examples are not tested automatically. Therefore,
I did manual testing and created a ticket to track this improvement:
KUDU-3478.
Manual testing has been done using the following configurations:
Ubuntu 18.04 x86_64: Python2.7
Ubuntu 18.04 x86_64: Python3.7
macOS Monterey M1(arm64): Python3.7
(I couldn't setup the Python2.7 interpreter properly on my mac)
STDOUT of the example:
kudu.Schema {
non_unique_key int32 NOT NULL
auto_incrementing_id int64 NOT NULL
int_val int32 NOT NULL
PRIMARY KEY (non_unique_key, auto_incrementing_id)
}
Demonstrating scanning ...
(non_unique_key: 3, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 3, auto_incrementing_id: 2, int_val: 1)
(non_unique_key: 3, auto_incrementing_id: 3, int_val: 2)
(non_unique_key: 4, auto_incrementing_id: 4, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 2, int_val: 1)
(non_unique_key: 1, auto_incrementing_id: 3, int_val: 2)
(non_unique_key: 2, auto_incrementing_id: 4, int_val: 0)
(non_unique_key: 2, auto_incrementing_id: 5, int_val: 1)
(non_unique_key: 2, auto_incrementing_id: 6, int_val: 2)
Scanned some row(s) WHERE non_unique_key = 1
(non_unique_key: 1, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 2, int_val: 1)
(non_unique_key: 1, auto_incrementing_id: 3, int_val: 2)
Demonstrating UPDATE ...
Updated row(s) WHERE non_unique_key = 1 AND int_val = 2 to int_val = 98
(non_unique_key: 3, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 3, auto_incrementing_id: 2, int_val: 1)
(non_unique_key: 3, auto_incrementing_id: 3, int_val: 2)
(non_unique_key: 4, auto_incrementing_id: 4, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 2, int_val: 1)
(non_unique_key: 1, auto_incrementing_id: 3, int_val: 98)
(non_unique_key: 2, auto_incrementing_id: 4, int_val: 0)
(non_unique_key: 2, auto_incrementing_id: 5, int_val: 1)
(non_unique_key: 2, auto_incrementing_id: 6, int_val: 2)
Updated row(s) WHERE non_unique_key = 2 to int_val = 99
(non_unique_key: 3, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 3, auto_incrementing_id: 2, int_val: 1)
(non_unique_key: 3, auto_incrementing_id: 3, int_val: 2)
(non_unique_key: 4, auto_incrementing_id: 4, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 2, int_val: 1)
(non_unique_key: 1, auto_incrementing_id: 3, int_val: 98)
(non_unique_key: 2, auto_incrementing_id: 4, int_val: 99)
(non_unique_key: 2, auto_incrementing_id: 5, int_val: 99)
(non_unique_key: 2, auto_incrementing_id: 6, int_val: 99)
Updated row(s) WHERE non_unique_key = 2 AND auto_incrementing_id = 5\
to int_val = 100
(non_unique_key: 3, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 3, auto_incrementing_id: 2, int_val: 1)
(non_unique_key: 3, auto_incrementing_id: 3, int_val: 2)
(non_unique_key: 4, auto_incrementing_id: 4, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 2, int_val: 1)
(non_unique_key: 1, auto_incrementing_id: 3, int_val: 98)
(non_unique_key: 2, auto_incrementing_id: 4, int_val: 99)
(non_unique_key: 2, auto_incrementing_id: 5, int_val: 100)
(non_unique_key: 2, auto_incrementing_id: 6, int_val: 99)
Demonstrating DELETE ...
Deleted row(s) WHERE non_unique_key = 3 AND int_val = 1
(non_unique_key: 3, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 3, auto_incrementing_id: 3, int_val: 2)
(non_unique_key: 4, auto_incrementing_id: 4, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 2, int_val: 1)
(non_unique_key: 1, auto_incrementing_id: 3, int_val: 98)
(non_unique_key: 2, auto_incrementing_id: 4, int_val: 99)
(non_unique_key: 2, auto_incrementing_id: 5, int_val: 100)
(non_unique_key: 2, auto_incrementing_id: 6, int_val: 99)
Deleted row(s) WHERE non_unique_key = 2
(non_unique_key: 3, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 3, auto_incrementing_id: 3, int_val: 2)
(non_unique_key: 4, auto_incrementing_id: 4, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 2, int_val: 1)
(non_unique_key: 1, auto_incrementing_id: 3, int_val: 98)
Deleted row(s) WHERE non_unique_key = 3 AND auto_incrementing_id = 3
(non_unique_key: 3, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 4, auto_incrementing_id: 4, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 1, int_val: 0)
(non_unique_key: 1, auto_incrementing_id: 2, int_val: 1)
(non_unique_key: 1, auto_incrementing_id: 3, int_val: 98)
Deleted the table
Done
Change-Id: I1dd862c4f26b3d79ec268727363fae3426135f52
Reviewed-on: http://gerrit.cloudera.org:8080/19874
Tested-by: Kudu Jenkins
Reviewed-by: Wenzhe Zhou <wz...@cloudera.com>
Reviewed-by: Attila Bukor <ab...@apache.org>
---
.../basic-python-example/non_unique_primary_key.py | 221 +++++++++++++++++++++
1 file changed, 221 insertions(+)
diff --git a/examples/python/basic-python-example/non_unique_primary_key.py b/examples/python/basic-python-example/non_unique_primary_key.py
new file mode 100644
index 000000000..7adaf4391
--- /dev/null
+++ b/examples/python/basic-python-example/non_unique_primary_key.py
@@ -0,0 +1,221 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import argparse
+
+import kudu
+from kudu import Schema
+from kudu.client import Partitioning
+
+# Pretty print scan results according to the schema
+def print_tuple(t):
+ print('(non_unique_key: {0}, {1}: {2}, int_val: {3})'.format(t[0],\
+ Schema.get_auto_incrementing_column_name(), t[1], t[2]))
+
+def print_rows(scanner):
+ scanner = scanner.open()
+ # The rows contains the auto-incrementing column. If one doesn't requre it, it can be
+ # discarded through a projection.
+ while scanner.has_more_rows():
+ for t in scanner.read_next_batch_tuples():
+ print_tuple(t)
+
+def update_rows(client, table, scanner, new_val):
+ # It's necessary to specify the entire set of key columns when updating a particular row.
+ # An auto-incrementing column is auto-populated at the server side, and one way to retrieve
+ # its values is scanning the table with a projection that includes the auto-incrementing column.
+ session = client.new_session()
+ while scanner.has_more_rows():
+ for t in scanner.read_next_batch_tuples():
+ op = table.new_update()
+ op['non_unique_key'] = t[0]
+ op[Schema.get_auto_incrementing_column_name()] = t[1]
+ op['int_val'] = new_val
+ session.apply(op)
+
+ try:
+ session.flush()
+ except kudu.KuduBadStatus:
+ print(session.get_pending_errors())
+
+def delete_rows(client, table, scanner):
+ # It's necessary to specify the entire set of key columns when updating a particular row.
+ # An auto-incrementing column is auto-populated at the server side, and one way to retrieve
+ # its values is scanning the table with a projection that includes the auto-incrementing column.
+ session = client.new_session()
+ while scanner.has_more_rows():
+ for t in scanner.read_next_batch_tuples():
+ op = table.new_delete()
+ op['non_unique_key'] = t[0]
+ op[Schema.get_auto_incrementing_column_name()] = t[1]
+ session.apply(op)
+
+ try:
+ session.flush()
+ except kudu.KuduBadStatus:
+ print(session.get_pending_errors())
+
+# Parse arguments
+parser = argparse.ArgumentParser(description='Basic Example for Kudu Python.')
+parser.add_argument('--masters', '-m', nargs='+', default='localhost',
+ help='The master address(es) to connect to Kudu.')
+parser.add_argument('--ports', '-p', nargs='+', default='7051',
+ help='The master server port(s) to connect to Kudu.')
+args = parser.parse_args()
+
+client = kudu.connect(host=args.masters, port=args.ports)
+
+builder = kudu.schema_builder()
+# Columns which are not uniquely identifiable can still be used as primary keys by
+# specifying them as non-unique primary key.
+builder.add_column('non_unique_key').type(kudu.int32).nullable(False).non_unique_primary_key()
+builder.add_column('int_val', type_=kudu.int32, nullable=False)
+schema = builder.build()
+
+# The schema stringification shows the presence of the auto-incrementing column,
+# and the resulting composite primary key.
+print(schema)
+
+partitioning = Partitioning().add_hash_partitions(column_names=['non_unique_key'], num_buckets=2)
+
+table_name = 'non_unique_key-example'
+if client.table_exists(table_name):
+ client.delete_table(table_name)
+client.create_table(table_name, schema, partitioning)
+
+table = client.table(table_name)
+session = client.new_session()
+stale_counter = 0
+num_rows = 10
+divisor = 3
+for i in range(num_rows):
+ if i % divisor == 0:
+ stale_counter += 1
+
+ op = table.new_insert()
+ # The auto-incrementing column is populated on the server-side automatically.
+ op['non_unique_key'] = stale_counter
+ op['int_val'] = i % divisor
+ session.apply(op)
+
+try:
+ session.flush()
+except kudu.KuduBadStatus:
+ print(session.get_pending_errors())
+
+print('Demonstrating scanning ...')
+scanner = table.scanner()
+print_rows(scanner)
+
+non_unique_key_equals = 1
+print('Scanned some row(s) WHERE non_unique_key = {0}'\
+ .format(non_unique_key_equals))
+scanner = table.scanner()
+scanner.add_predicate(table['non_unique_key'] == non_unique_key_equals).open()
+print_rows(scanner)
+
+print('Demonstrating UPDATE ...')
+# Updating based upon a predicate on a non-unique PK and on a non-PK column
+non_unique_key_equals = 1
+int_val_equals = 2
+new_val = 98
+scanner = table.scanner()
+scanner.add_predicate(table['non_unique_key'] == non_unique_key_equals)\
+ .add_predicate(table['int_val'] == int_val_equals)\
+ .open()
+update_rows(client, table, scanner, new_val)
+
+print('Updated row(s) WHERE non_unique_key = {0} AND int_val = {1} to int_val = {2}'\
+ .format(non_unique_key_equals, int_val_equals, new_val))
+scanner = table.scanner()
+print_rows(scanner)
+
+# Updating based upon a predicate on a non-unique PK
+non_unique_key_equals = 2
+new_val = 99
+scanner = table.scanner()
+scanner.add_predicate(table['non_unique_key'] == non_unique_key_equals).open()
+update_rows(client, table, scanner, new_val)
+
+print('Updated row(s) WHERE non_unique_key = {0} to int_val = {1}'\
+ .format(non_unique_key_equals, new_val))
+scanner = table.scanner()
+print_rows(scanner)
+
+# Updating based upon a predicate on a non-unique PK and on the auto-incrementing column
+non_unique_key_equals = 2
+auto_incrementing_counter_val = 5
+new_val = 100
+scanner = table.scanner()
+scanner.add_predicate(table['non_unique_key'] == non_unique_key_equals)\
+ .add_predicate(table[Schema.get_auto_incrementing_column_name()] ==\
+ auto_incrementing_counter_val)\
+ .open()
+update_rows(client, table, scanner, new_val)
+
+print('Updated row(s) WHERE non_unique_key = {0} AND {1} = {2} to int_val = {3}'\
+ .format(non_unique_key_equals, Schema.get_auto_incrementing_column_name(),\
+ auto_incrementing_counter_val, new_val))
+scanner = table.scanner()
+print_rows(scanner)
+
+print('Demonstrating DELETE ...')
+# Deleting based upon a predicate on a non-unique PK and on a non-PK column
+non_unique_key_equals = 3
+int_val_equals = 1
+scanner = table.scanner()
+scanner.add_predicate(table['non_unique_key'] == non_unique_key_equals)\
+ .add_predicate(table['int_val'] == int_val_equals)\
+ .open()
+delete_rows(client, table, scanner)
+
+print('Deleted row(s) WHERE non_unique_key = {0} AND int_val = {1}'\
+ .format(non_unique_key_equals, int_val_equals))
+scanner = table.scanner()
+print_rows(scanner)
+
+# Deleting based upon a predicate on a non-unique PK
+non_unique_key_equals = 2
+scanner = table.scanner()
+scanner.add_predicate(table['non_unique_key'] == non_unique_key_equals)\
+ .open()
+delete_rows(client, table, scanner)
+
+print('Deleted row(s) WHERE non_unique_key = {0}'\
+ .format(non_unique_key_equals, int_val_equals))
+scanner = table.scanner()
+print_rows(scanner)
+
+# Deleting based upon a predicate on a non-unique PK and on the auto-incrementing column
+non_unique_key_equals = 3
+auto_incrementing_counter_val = 3
+scanner = table.scanner()
+scanner.add_predicate(table['non_unique_key'] == non_unique_key_equals)\
+ .add_predicate(table[Schema.get_auto_incrementing_column_name()] ==\
+ auto_incrementing_counter_val)\
+ .open()
+delete_rows(client, table, scanner)
+
+print('Deleted row(s) WHERE non_unique_key = {0} AND {1} = {2}'\
+ .format(non_unique_key_equals, Schema.get_auto_incrementing_column_name(),\
+ auto_incrementing_counter_val))
+scanner = table.scanner()
+print_rows(scanner)
+
+client.delete_table(table_name)
+print('Deleted the table')
+print('Done')
\ No newline at end of file
[kudu] 03/03: [Client] Add query id to trace the whole query process
Posted by la...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
laiyingchun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 570936fa84f3d652a88830425e29cc486526d72a
Author: xinghuayu007 <14...@qq.com>
AuthorDate: Mon Feb 20 15:58:07 2023 +0800
[Client] Add query id to trace the whole query process
This is a follow up patch of 834de7fccdb5faadb2ca9e1d1e07d4c7882ae0fa.
Query id is used to trace a query from upstream applications to Kudu
for troubleshooting and debugging.
Change-Id: Iddbd348e766bad1b5648b4091635679319b1e2fd
Reviewed-on: http://gerrit.cloudera.org:8080/19518
Reviewed-by: Yingchun Lai <la...@apache.org>
Tested-by: Yingchun Lai <la...@apache.org>
---
.../kudu/client/AbstractKuduScannerBuilder.java | 16 +++++++
.../org/apache/kudu/client/AsyncKuduScanner.java | 27 ++++++++++-
.../java/org/apache/kudu/client/KuduScanToken.java | 4 ++
.../java/org/apache/kudu/client/KuduScanner.java | 2 +-
.../org/apache/kudu/client/TestKuduScanner.java | 39 ++++++++++++++++
.../java/org/apache/kudu/client/TestScanToken.java | 53 ++++++++++++++++++++++
6 files changed, 139 insertions(+), 2 deletions(-)
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AbstractKuduScannerBuilder.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AbstractKuduScannerBuilder.java
index 2c9b4aa36..db843efca 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AbstractKuduScannerBuilder.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AbstractKuduScannerBuilder.java
@@ -58,6 +58,7 @@ public abstract class AbstractKuduScannerBuilder
long scanRequestTimeout;
ReplicaSelection replicaSelection = ReplicaSelection.LEADER_ONLY;
long keepAlivePeriodMs = AsyncKuduClient.DEFAULT_KEEP_ALIVE_PERIOD_MS;
+ String queryId = "";
AbstractKuduScannerBuilder(AsyncKuduClient client, KuduTable table) {
this.client = client;
@@ -412,5 +413,20 @@ public abstract class AbstractKuduScannerBuilder
return (S) this;
}
+ /**
+ * Set a query id for the scan to trace the whole scanning process.
+ * Query id is posted by the user or generated automatically by the
+ * client library code. It is used to trace the whole query process
+ * for debugging.
+ *
+ * @param queryId query id to trace a query.
+ * @return this instance
+ */
+ @SuppressWarnings("unchecked")
+ public S setQueryId(String queryId) {
+ this.queryId = queryId;
+ return (S) this;
+ }
+
public abstract T build();
}
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
index 4929dd4d8..a688224ab 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
@@ -38,6 +38,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
import com.google.common.collect.ImmutableList;
@@ -284,6 +285,8 @@ public final class AsyncKuduScanner {
final long scanRequestTimeout;
+ private String queryId;
+
/**
* The prefetching result is cached in memory. This atomic reference is used to avoid
* two concurrent prefetchings occur and the latest one overrides the previous one.
@@ -399,6 +402,27 @@ public final class AsyncKuduScanner {
}
}
+ AsyncKuduScanner(AsyncKuduClient client, KuduTable table, List<String> projectedNames,
+ List<Integer> projectedIndexes, ReadMode readMode, boolean isFaultTolerant,
+ long scanRequestTimeout,
+ Map<String, KuduPredicate> predicates, long limit,
+ boolean cacheBlocks, boolean prefetching,
+ byte[] startPrimaryKey, byte[] endPrimaryKey,
+ long startTimestamp, long htTimestamp,
+ int batchSizeBytes, PartitionPruner pruner,
+ ReplicaSelection replicaSelection, long keepAlivePeriodMs, String queryId) {
+ this(
+ client, table, projectedNames, projectedIndexes, readMode, isFaultTolerant,
+ scanRequestTimeout, predicates, limit, cacheBlocks, prefetching, startPrimaryKey,
+ endPrimaryKey, startTimestamp, htTimestamp, batchSizeBytes,
+ pruner, replicaSelection, keepAlivePeriodMs);
+ if (queryId.isEmpty()) {
+ this.queryId = UUID.randomUUID().toString().replace("-", "");
+ } else {
+ this.queryId = queryId;
+ }
+ }
+
/**
* Generates and returns a ColumnSchema for the virtual IS_DELETED column.
* The column name is generated to ensure there is never a collision.
@@ -1214,6 +1238,7 @@ public final class AsyncKuduScanner {
default:
throw new RuntimeException("unreachable!");
}
+ builder.setQueryId(UnsafeByteOperations.unsafeWrap(queryId.getBytes(UTF_8)));
return builder.build();
}
@@ -1326,7 +1351,7 @@ public final class AsyncKuduScanner {
client, table, projectedColumnNames, projectedColumnIndexes, readMode, isFaultTolerant,
scanRequestTimeout, predicates, limit, cacheBlocks, prefetching, lowerBoundPrimaryKey,
upperBoundPrimaryKey, startTimestamp, htTimestamp, batchSizeBytes,
- PartitionPruner.create(this), replicaSelection, keepAlivePeriodMs);
+ PartitionPruner.create(this), replicaSelection, keepAlivePeriodMs, queryId);
}
}
}
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java
index e86408951..3a5878a15 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanToken.java
@@ -366,6 +366,9 @@ public class KuduScanToken implements Comparable<KuduScanToken> {
if (message.hasKeepAlivePeriodMs()) {
builder.keepAlivePeriodMs(message.getKeepAlivePeriodMs());
}
+ if (message.hasQueryId()) {
+ builder.setQueryId(message.getQueryId());
+ }
return builder;
}
@@ -724,6 +727,7 @@ public class KuduScanToken implements Comparable<KuduScanToken> {
builder.setTabletMetadata(tabletMetadataPB);
}
}
+ builder.setQueryId(queryId);
tokens.add(new KuduScanToken(keyRange.getTablet(), builder.build()));
}
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanner.java b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanner.java
index 6e16a7be9..02142630b 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanner.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduScanner.java
@@ -224,7 +224,7 @@ public class KuduScanner implements Iterable<RowResult> {
client, table, projectedColumnNames, projectedColumnIndexes, readMode, isFaultTolerant,
scanRequestTimeout, predicates, limit, cacheBlocks, prefetching, lowerBoundPrimaryKey,
upperBoundPrimaryKey, startTimestamp, htTimestamp, batchSizeBytes,
- PartitionPruner.create(this), replicaSelection, keepAlivePeriodMs));
+ PartitionPruner.create(this), replicaSelection, keepAlivePeriodMs, queryId));
}
}
}
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduScanner.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduScanner.java
index 54e7d6577..8ea7a1e08 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduScanner.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduScanner.java
@@ -233,6 +233,45 @@ public class TestKuduScanner {
}
}
+ @Test(timeout = 100000)
+ public void testScanWithQueryId() throws Exception {
+ KuduTable table = client.createTable(tableName, getBasicSchema(), getBasicCreateTableOptions());
+ DataGenerator generator = new DataGenerator.DataGeneratorBuilder()
+ .random(RandomUtils.getRandom())
+ .build();
+ KuduSession session = client.newSession();
+ int numRows = 10;
+ for (int i = 0; i < numRows; i++) {
+ Insert insert = table.newInsert();
+ PartialRow row = insert.getRow();
+ generator.randomizeRow(row);
+ session.apply(insert);
+ }
+ // Scan with specified query id.
+ {
+ int rowsScanned = 0;
+ KuduScanner scanner = client.newScannerBuilder(table)
+ .batchSizeBytes(100)
+ .setQueryId("request-id-for-test")
+ .build();
+ while (scanner.hasMoreRows()) {
+ rowsScanned += scanner.nextRows().getNumRows();
+ }
+ assertEquals(numRows, rowsScanned);
+ }
+ // Scan with default query id.
+ {
+ int rowsScanned = 0;
+ KuduScanner scanner = client.newScannerBuilder(table)
+ .batchSizeBytes(100)
+ .build();
+ while (scanner.hasMoreRows()) {
+ rowsScanned += scanner.nextRows().getNumRows();
+ }
+ assertEquals(numRows, rowsScanned);
+ }
+ }
+
@Test(timeout = 100000)
public void testOpenScanWithDroppedPartition() throws Exception {
// Create a table with 2 range partitions.
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestScanToken.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestScanToken.java
index 4ed897085..22aedbd6c 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestScanToken.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestScanToken.java
@@ -142,6 +142,59 @@ public class TestScanToken {
}
}
+ @Test
+ public void testScanTokenWithQueryId() throws Exception {
+ // Prepare the table for testing.
+ Schema schema = createManyStringsSchema();
+ CreateTableOptions createOptions = new CreateTableOptions();
+ final int buckets = 8;
+ createOptions.addHashPartitions(ImmutableList.of("key"), buckets);
+ client.createTable(testTableName, schema, createOptions);
+
+ KuduSession session = client.newSession();
+ KuduTable table = client.openTable(testTableName);
+ final int totalRows = 100;
+ for (int i = 0; i < totalRows; i++) {
+ Insert insert = table.newInsert();
+ PartialRow row = insert.getRow();
+ row.addString("key", String.format("key_%02d", i));
+ row.addString("c1", "c1_" + i);
+ row.addString("c2", "c2_" + i);
+ assertEquals(session.apply(insert).hasRowError(), false);
+ }
+ // Scan with specified query id.
+ {
+ int rowsScanned = 0;
+ KuduScanToken.KuduScanTokenBuilder tokenBuilder = client.newScanTokenBuilder(table);
+ tokenBuilder.setProjectedColumnIndexes(ImmutableList.of());
+ tokenBuilder.setQueryId("query-id-for-test");
+ List<KuduScanToken> tokens = tokenBuilder.build();
+ assertEquals(buckets, tokens.size());
+ for (int i = 0; i < tokens.size(); i++) {
+ KuduScanner scanner = tokens.get(i).intoScanner(client);
+ while (scanner.hasMoreRows()) {
+ rowsScanned += scanner.nextRows().getNumRows();
+ }
+ }
+ assertEquals(totalRows, rowsScanned);
+ }
+ // Scan with default query id.
+ {
+ int rowsScanned = 0;
+ KuduScanToken.KuduScanTokenBuilder tokenBuilder = client.newScanTokenBuilder(table);
+ tokenBuilder.setProjectedColumnIndexes(ImmutableList.of());
+ List<KuduScanToken> tokens = tokenBuilder.build();
+ assertEquals(buckets, tokens.size());
+ for (int i = 0; i < tokens.size(); i++) {
+ KuduScanner scanner = tokens.get(i).intoScanner(client);
+ while (scanner.hasMoreRows()) {
+ rowsScanned += scanner.nextRows().getNumRows();
+ }
+ }
+ assertEquals(totalRows, rowsScanned);
+ }
+ }
+
/**
* Regression test for KUDU-3349
*/