You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by to...@apache.org on 2016/12/15 03:59:01 UTC
kudu git commit: KUDU-1806. java: fetching scan tokens should fetch
larger batches
Repository: kudu
Updated Branches:
refs/heads/master 4e4ea7982 -> 01e6871bc
KUDU-1806. java: fetching scan tokens should fetch larger batches
This changes the number of tablets fetched in a single GetTableLocations
RPC from 10 to 1000 for the case of scans or scan token generation. On a
stress test on 200 nodes with 40 concurrent query streams, this
substantially reduced the time spent in scan token generation by the
Impala planner.
This keeps the existing batch size (10) for cases where the client is
looking up a tablet for the purposes of a normal operation, but extends
it to 1000 for cases where we are explicitly fetching a range of tablet
locations. The aim here is that this will not increase network traffic
or master load for the case of random write workloads, but will increase
performance for "query planner" type workloads.
Although this will slightly increase the amount of work done by a
GetTableLocations RPC, my guess is that the majority of the RPC cost is
dominated by fixed per-RPC costs and not the linear cost based on the
number of tablets. This is especially true when taking into account the
typical RTT within a large/busy cluster (~1ms). So, it is a lot cheaper,
both in wall clock and total resources consumed, to process one larger
RPC rather than tens or hundreds of small ones.
This test also modifies the scan token generation test case to set the
fetch size down to a low value. This ensures that the code path to go
back and fetch more locations is still exercised, rather than always
fetching all of the tablets in one RPC.
Change-Id: I46260a96dfd0847f70146496e48c2766b8e17ea9
Reviewed-on: http://gerrit.cloudera.org:8080/5498
Reviewed-by: Dan Burkert <da...@apache.org>
Tested-by: Kudu Jenkins
Reviewed-by: Jean-Daniel Cryans <jd...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/01e6871b
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/01e6871b
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/01e6871b
Branch: refs/heads/master
Commit: 01e6871bca38ea2bed5c17290d43739a0c88094e
Parents: 4e4ea79
Author: Todd Lipcon <to...@apache.org>
Authored: Wed Dec 14 14:33:15 2016 +0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Thu Dec 15 03:53:27 2016 +0000
----------------------------------------------------------------------
.../org/apache/kudu/client/AsyncKuduClient.java | 68 ++++++++++++++------
.../kudu/client/GetTableLocationsRequest.java | 7 +-
.../java/org/apache/kudu/client/KuduTable.java | 8 ++-
.../apache/kudu/client/TableLocationsCache.java | 5 +-
.../apache/kudu/client/TestAsyncKuduClient.java | 11 ++--
.../org/apache/kudu/client/TestKuduClient.java | 61 ++++++++++--------
6 files changed, 107 insertions(+), 53 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/01e6871b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
index 78c190b..fb478f9 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
@@ -121,7 +121,18 @@ public class AsyncKuduClient implements AutoCloseable {
public static final long DEFAULT_OPERATION_TIMEOUT_MS = 30000;
public static final long DEFAULT_SOCKET_READ_TIMEOUT_MS = 10000;
private static final long MAX_RPC_ATTEMPTS = 100;
- static final int MAX_RETURNED_TABLE_LOCATIONS = 10;
+
+ /**
+ * The number of tablets to fetch from the master in a round trip when performing
+ * a lookup of a single partition (e.g. for a write), or re-looking-up a tablet with
+ * stale information.
+ */
+ static int FETCH_TABLETS_PER_POINT_LOOKUP = 10;
+ /**
+ * The number of tablets to fetch from the master when looking up a range of
+ * tablets.
+ */
+ static int FETCH_TABLETS_PER_RANGE_LOOKUP = 1000;
private final ClientSocketChannelFactory channelFactory;
@@ -723,7 +734,7 @@ public class AsyncKuduClient implements AutoCloseable {
Callback<Deferred<R>, Master.GetTableLocationsResponsePB> cb = new RetryRpcCB<>(request);
Callback<Deferred<R>, Exception> eb = new RetryRpcErrback<>(request);
Deferred<Master.GetTableLocationsResponsePB> returnedD =
- locateTablet(request.getTable(), partitionKey, request);
+ locateTablet(request.getTable(), partitionKey, FETCH_TABLETS_PER_POINT_LOOKUP, request);
return AsyncUtil.addCallbacksDeferring(returnedD, cb, eb);
}
@@ -733,7 +744,7 @@ public class AsyncKuduClient implements AutoCloseable {
* <p>
* Use {@code AsyncUtil.addCallbacksDeferring} to add this as the callback and
* {@link AsyncKuduClient.RetryRpcErrback} as the "errback" to the {@code Deferred}
- * returned by {@link #locateTablet(KuduTable, byte[], KuduRpc)}.
+ * returned by {@link #locateTablet(KuduTable, byte[], int, KuduRpc)}.
* @param <R> RPC's return type.
* @param <D> Previous query's return type, which we don't use, but need to specify in order to
* tie it all together.
@@ -763,7 +774,7 @@ public class AsyncKuduClient implements AutoCloseable {
* <p>
* Use {@code AsyncUtil.addCallbacksDeferring} to add this as the "errback" and
* {@link RetryRpcCB} as the callback to the {@code Deferred} returned by
- * {@link #locateTablet(KuduTable, byte[], KuduRpc)}.
+ * {@link #locateTablet(KuduTable, byte[], int, KuduRpc)}.
* @see #delayedSendRpcToTablet(KuduRpc, KuduException)
* @param <R> The type of the original RPC.
*/
@@ -1001,11 +1012,13 @@ public class AsyncKuduClient implements AutoCloseable {
* Sends a getTableLocations RPC to the master to find the table's tablets.
* @param table table to lookup
* @param partitionKey can be null, if not we'll find the exact tablet that contains it
+ * @param fetchBatchSize the number of tablets to fetch per round trip from the master
* @param parentRpc RPC that prompted a master lookup, can be null
* @return Deferred to track the progress
*/
private Deferred<Master.GetTableLocationsResponsePB> locateTablet(KuduTable table,
byte[] partitionKey,
+ int fetchBatchSize,
KuduRpc<?> parentRpc) {
boolean hasPermit = acquireMasterLookupPermit();
String tableId = table.getTableId();
@@ -1029,7 +1042,7 @@ public class AsyncKuduClient implements AutoCloseable {
} else {
// Leave the end of the partition key range empty in order to pre-fetch tablet locations.
GetTableLocationsRequest rpc =
- new GetTableLocationsRequest(masterTable, partitionKey, null, tableId);
+ new GetTableLocationsRequest(masterTable, partitionKey, null, tableId, fetchBatchSize);
if (parentRpc != null) {
rpc.setTimeoutMillis(parentRpc.deadlineTracker.getMillisBeforeDeadline());
rpc.setParentRpc(parentRpc);
@@ -1038,7 +1051,7 @@ public class AsyncKuduClient implements AutoCloseable {
}
d = sendRpcToTablet(rpc);
}
- d.addCallback(new MasterLookupCB(table, partitionKey));
+ d.addCallback(new MasterLookupCB(table, partitionKey, fetchBatchSize));
if (hasPermit) {
d.addBoth(new ReleaseMasterLookupPermit<Master.GetTableLocationsResponsePB>());
}
@@ -1083,6 +1096,7 @@ public class AsyncKuduClient implements AutoCloseable {
* @param startPartitionKey where to start in the table, pass null to start at the beginning
* @param endPartitionKey where to stop in the table, pass null to get all the tablets until the
* end of the table
+ * @param fetchBatchSize the number of tablets to fetch per round trip from the master
* @param deadline deadline in milliseconds for this method to finish
* @return a list of the tablets in the table, which can be queried for metadata about
* each tablet
@@ -1091,13 +1105,15 @@ public class AsyncKuduClient implements AutoCloseable {
List<LocatedTablet> syncLocateTable(KuduTable table,
byte[] startPartitionKey,
byte[] endPartitionKey,
+ int fetchBatchSize,
long deadline) throws Exception {
- return locateTable(table, startPartitionKey, endPartitionKey, deadline).join();
+ return locateTable(table, startPartitionKey, endPartitionKey, fetchBatchSize, deadline).join();
}
private Deferred<List<LocatedTablet>> loopLocateTable(final KuduTable table,
final byte[] startPartitionKey,
final byte[] endPartitionKey,
+ final int fetchBatchSize,
final List<LocatedTablet> ret,
final DeadlineTracker deadlineTracker) {
// We rely on the keys initially not being empty.
@@ -1138,11 +1154,12 @@ public class AsyncKuduClient implements AutoCloseable {
// When lookup completes, the tablet (or non-covered range) for the next
// partition key will be located and added to the client's cache.
final byte[] lookupKey = partitionKey;
- return locateTablet(table, key, null).addCallbackDeferring(
+ return locateTablet(table, key, fetchBatchSize, null).addCallbackDeferring(
new Callback<Deferred<List<LocatedTablet>>, GetTableLocationsResponsePB>() {
@Override
public Deferred<List<LocatedTablet>> call(GetTableLocationsResponsePB resp) {
- return loopLocateTable(table, lookupKey, endPartitionKey, ret, deadlineTracker);
+ return loopLocateTable(table, lookupKey, endPartitionKey, fetchBatchSize,
+ ret, deadlineTracker);
}
@Override
@@ -1162,6 +1179,7 @@ public class AsyncKuduClient implements AutoCloseable {
* @param startPartitionKey where to start in the table, pass null to start at the beginning
* @param endPartitionKey where to stop in the table, pass null to get all the tablets until the
* end of the table
+ * @param fetchBatchSize the number of tablets to fetch per round trip from the master
* @param deadline max time spent in milliseconds for the deferred result of this method to
* get called back, if deadline is reached, the deferred result will get erred back
* @return a deferred object that yields a list of the tablets in the table, which can be queried
@@ -1170,11 +1188,13 @@ public class AsyncKuduClient implements AutoCloseable {
Deferred<List<LocatedTablet>> locateTable(final KuduTable table,
final byte[] startPartitionKey,
final byte[] endPartitionKey,
+ int fetchBatchSize,
long deadline) {
final List<LocatedTablet> ret = Lists.newArrayList();
final DeadlineTracker deadlineTracker = new DeadlineTracker();
deadlineTracker.setDeadline(deadline);
- return loopLocateTable(table, startPartitionKey, endPartitionKey, ret, deadlineTracker);
+ return loopLocateTable(table, startPartitionKey, endPartitionKey, fetchBatchSize,
+ ret, deadlineTracker);
}
/**
@@ -1256,10 +1276,12 @@ public class AsyncKuduClient implements AutoCloseable {
Master.GetTableLocationsResponsePB> {
final KuduTable table;
private final byte[] partitionKey;
+ private final int requestedBatchSize;
- MasterLookupCB(KuduTable table, byte[] partitionKey) {
+ MasterLookupCB(KuduTable table, byte[] partitionKey, int requestedBatchSize) {
this.table = table;
this.partitionKey = partitionKey;
+ this.requestedBatchSize = requestedBatchSize;
}
public Object call(final GetTableLocationsResponsePB response) {
@@ -1276,6 +1298,7 @@ public class AsyncKuduClient implements AutoCloseable {
try {
discoverTablets(table,
partitionKey,
+ requestedBatchSize,
response.getTabletLocationsList(),
response.getTtlMillis());
} catch (KuduException e) {
@@ -1313,12 +1336,15 @@ public class AsyncKuduClient implements AutoCloseable {
* Makes discovered tablet locations visible in the clients caches.
* @param table the table which the locations belong to
* @param requestPartitionKey the partition key of the table locations request
+ * @param requestedBatchSize the number of tablet locations requested from the master in the
+ * original request
* @param locations the discovered locations
* @param ttl the ttl of the locations
*/
@VisibleForTesting
void discoverTablets(KuduTable table,
byte[] requestPartitionKey,
+ int requestedBatchSize,
List<Master.TabletLocationsPB> locations,
long ttl) throws KuduException {
String tableId = table.getTableId();
@@ -1371,7 +1397,7 @@ public class AsyncKuduClient implements AutoCloseable {
// Give the locations to the tablet location cache for the table, so that it
// can cache them and discover non-covered ranges.
- locationsCache.cacheTabletLocations(tablets, requestPartitionKey, ttl);
+ locationsCache.cacheTabletLocations(tablets, requestPartitionKey, requestedBatchSize, ttl);
// Now test if we found the tablet we were looking for. If so, RetryRpcCB will retry the RPC
// right away. If not, we throw an exception that RetryRpcErrback will understand as needing to
@@ -1407,17 +1433,23 @@ public class AsyncKuduClient implements AutoCloseable {
Deferred<LocatedTablet> getTabletLocation(final KuduTable table,
final byte[] partitionKey,
long deadline) {
- // Locate the tablets at the partition key by locating all tablets between
- // the partition key (inclusive), and the incremented partition key (exclusive).
- Deferred<List<LocatedTablet>> locatedTablets;
+ // Locate the tablet at the partition key by locating tablets between
+ // the partition key (inclusive), and the incremented partition key (exclusive).
+ // We expect this to return at most a single tablet (checked below).
+ byte[] startPartitionKey;
+ byte[] endPartitionKey;
if (partitionKey.length == 0) {
- locatedTablets = locateTable(table, null, new byte[] { 0x00 }, deadline);
+ startPartitionKey = null;
+ endPartitionKey = new byte[] { 0x00 };
} else {
- locatedTablets = locateTable(table, partitionKey,
- Arrays.copyOf(partitionKey, partitionKey.length + 1), deadline);
+ startPartitionKey = partitionKey;
+ endPartitionKey = Arrays.copyOf(partitionKey, partitionKey.length + 1);
}
+ Deferred<List<LocatedTablet>> locatedTablets = locateTable(
+ table, startPartitionKey, endPartitionKey, FETCH_TABLETS_PER_POINT_LOOKUP, deadline);
+
// Then pick out the single tablet result from the list.
return locatedTablets.addCallbackDeferring(
new Callback<Deferred<LocatedTablet>, List<LocatedTablet>>() {
http://git-wip-us.apache.org/repos/asf/kudu/blob/01e6871b/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableLocationsRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableLocationsRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableLocationsRequest.java
index 75857fa..e7d4637 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableLocationsRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableLocationsRequest.java
@@ -35,9 +35,11 @@ class GetTableLocationsRequest extends KuduRpc<Master.GetTableLocationsResponseP
private final byte[] startPartitionKey;
private final byte[] endKey;
private final String tableId;
+ private final int maxReturnedLocations;
GetTableLocationsRequest(KuduTable table, byte[] startPartitionKey,
- byte[] endPartitionKey, String tableId) {
+ byte[] endPartitionKey, String tableId,
+ int maxReturnedLocations) {
super(table);
if (startPartitionKey != null && endPartitionKey != null &&
Bytes.memcmp(startPartitionKey, endPartitionKey) > 0) {
@@ -47,6 +49,7 @@ class GetTableLocationsRequest extends KuduRpc<Master.GetTableLocationsResponseP
this.startPartitionKey = startPartitionKey;
this.endKey = endPartitionKey;
this.tableId = tableId;
+ this.maxReturnedLocations = maxReturnedLocations;
}
@Override
@@ -83,7 +86,7 @@ class GetTableLocationsRequest extends KuduRpc<Master.GetTableLocationsResponseP
if (endKey != null) {
builder.setPartitionKeyEnd(ZeroCopyLiteralByteString.wrap(endKey));
}
- builder.setMaxReturnedLocations(AsyncKuduClient.MAX_RETURNED_TABLE_LOCATIONS);
+ builder.setMaxReturnedLocations(maxReturnedLocations);
return toChannelBuffer(header, builder.build());
}
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/01e6871b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduTable.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduTable.java b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduTable.java
index b990369..b12307c 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduTable.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduTable.java
@@ -168,7 +168,9 @@ public class KuduTable {
public Deferred<List<LocatedTablet>> asyncGetTabletsLocations(byte[] startKey,
byte[] endKey,
long deadline) {
- return client.locateTable(this, startKey, endKey, deadline);
+ return client.locateTable(this, startKey, endKey,
+ AsyncKuduClient.FETCH_TABLETS_PER_RANGE_LOOKUP,
+ deadline);
}
/**
@@ -202,7 +204,9 @@ public class KuduTable {
public List<LocatedTablet> getTabletsLocations(byte[] startKey,
byte[] endKey,
long deadline) throws Exception {
- return client.syncLocateTable(this, startKey, endKey, deadline);
+ return client.syncLocateTable(this, startKey, endKey,
+ AsyncKuduClient.FETCH_TABLETS_PER_RANGE_LOOKUP,
+ deadline);
}
/**
http://git-wip-us.apache.org/repos/asf/kudu/blob/01e6871b/java/kudu-client/src/main/java/org/apache/kudu/client/TableLocationsCache.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/TableLocationsCache.java b/java/kudu-client/src/main/java/org/apache/kudu/client/TableLocationsCache.java
index 25678ce..12400d4 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/TableLocationsCache.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/TableLocationsCache.java
@@ -89,10 +89,13 @@ class TableLocationsCache {
*
* @param tablets the discovered tablets to cache
* @param requestPartitionKey the lookup partition key
+ * @param requestedBatchSize the number of tablet locations requested from the master in the
+ * original request
* @param ttl the time in milliseconds that the tablets may be cached for
*/
public void cacheTabletLocations(List<RemoteTablet> tablets,
byte[] requestPartitionKey,
+ int requestedBatchSize,
long ttl) {
long deadline = System.nanoTime() + ttl * TimeUnit.MILLISECONDS.toNanos(1);
if (requestPartitionKey == null) {
@@ -166,7 +169,7 @@ class TableLocationsCache {
}
if (lastUpperBound.length > 0 &&
- tablets.size() < AsyncKuduClient.MAX_RETURNED_TABLE_LOCATIONS) {
+ tablets.size() < requestedBatchSize) {
// There is a non-covered range between the last tablet and the end of the
// partition key space, such as F.
newEntries.add(
http://git-wip-us.apache.org/repos/asf/kudu/blob/01e6871b/java/kudu-client/src/test/java/org/apache/kudu/client/TestAsyncKuduClient.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestAsyncKuduClient.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestAsyncKuduClient.java
index 302b2cc..3502909 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestAsyncKuduClient.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestAsyncKuduClient.java
@@ -123,6 +123,7 @@ public class TestAsyncKuduClient extends BaseKuduTest {
@Test
public void testBadHostnames() throws Exception {
String badHostname = "some-unknown-host-hopefully";
+ final int requestBatchSize = 10;
// Test that a bad hostname for the master makes us error out quickly.
AsyncKuduClient invalidClient = new AsyncKuduClient.AsyncKuduClientBuilder(badHostname).build();
@@ -153,7 +154,7 @@ public class TestAsyncKuduClient extends BaseKuduTest {
try {
KuduTable badTable = new KuduTable(client, "Invalid table name",
"Invalid table ID", null, null);
- client.discoverTablets(badTable, null, tabletLocations, 1000);
+ client.discoverTablets(badTable, null, requestBatchSize, tabletLocations, 1000);
fail("This should have failed quickly");
} catch (NonRecoverableException ex) {
assertTrue(ex.getMessage().contains(badHostname));
@@ -162,6 +163,7 @@ public class TestAsyncKuduClient extends BaseKuduTest {
@Test
public void testNoLeader() throws Exception {
+ final int requestBatchSize = 10;
CreateTableOptions options = getBasicCreateTableOptions();
KuduTable table = createTable(
"testNoLeader-" + System.currentTimeMillis(),
@@ -169,8 +171,9 @@ public class TestAsyncKuduClient extends BaseKuduTest {
options);
// Lookup the current locations so that we can pass some valid information to discoverTablets.
- List<LocatedTablet> tablets =
- client.locateTable(table, null, null, DEFAULT_SLEEP).join(DEFAULT_SLEEP);
+ List<LocatedTablet> tablets = client
+ .locateTable(table, null, null, requestBatchSize, DEFAULT_SLEEP)
+ .join(DEFAULT_SLEEP);
LocatedTablet tablet = tablets.get(0);
LocatedTablet.Replica leader = tablet.getLeaderReplica();
@@ -183,7 +186,7 @@ public class TestAsyncKuduClient extends BaseKuduTest {
"master", leader.getRpcHost(), leader.getRpcPort(), Metadata.RaftPeerPB.Role.FOLLOWER));
tabletLocations.add(tabletPb.build());
try {
- client.discoverTablets(table, new byte[0], tabletLocations, 1000);
+ client.discoverTablets(table, new byte[0], requestBatchSize, tabletLocations, 1000);
fail("discoverTablets should throw an exception if there's no leader");
} catch (NoLeaderFoundException ex) {
// Expected.
http://git-wip-us.apache.org/repos/asf/kudu/blob/01e6871b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
index 64951b2..b72f8fc 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
@@ -513,37 +513,46 @@ public class TestKuduClient extends BaseKuduTest {
*/
@Test
public void testScanTokens() throws Exception {
- Schema schema = createManyStringsSchema();
- CreateTableOptions createOptions = new CreateTableOptions();
- createOptions.addHashPartitions(ImmutableList.of("key"), 8);
+ int saveFetchTablets = AsyncKuduClient.FETCH_TABLETS_PER_RANGE_LOOKUP;
+ try {
+ // For this test, make sure that we cover the case that not all tablets
+ // are returned in a single batch.
+ AsyncKuduClient.FETCH_TABLETS_PER_RANGE_LOOKUP = 4;
- PartialRow splitRow = schema.newPartialRow();
- splitRow.addString("key", "key_50");
- createOptions.addSplitRow(splitRow);
+ Schema schema = createManyStringsSchema();
+ CreateTableOptions createOptions = new CreateTableOptions();
+ createOptions.addHashPartitions(ImmutableList.of("key"), 8);
- syncClient.createTable(tableName, schema, createOptions);
+ PartialRow splitRow = schema.newPartialRow();
+ splitRow.addString("key", "key_50");
+ createOptions.addSplitRow(splitRow);
- KuduSession session = syncClient.newSession();
- session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);
- KuduTable table = syncClient.openTable(tableName);
- for (int i = 0; i < 100; 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);
- session.apply(insert);
- }
- session.flush();
+ syncClient.createTable(tableName, schema, createOptions);
- KuduScanToken.KuduScanTokenBuilder tokenBuilder = syncClient.newScanTokenBuilder(table);
- tokenBuilder.setProjectedColumnIndexes(ImmutableList.<Integer>of());
- List<KuduScanToken> tokens = tokenBuilder.build();
- assertEquals(16, tokens.size());
+ KuduSession session = syncClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);
+ KuduTable table = syncClient.openTable(tableName);
+ for (int i = 0; i < 100; 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);
+ session.apply(insert);
+ }
+ session.flush();
- for (KuduScanToken token : tokens) {
- // Sanity check to make sure the debug printing does not throw.
- LOG.debug(KuduScanToken.stringifySerializedToken(token.serialize(), syncClient));
+ KuduScanToken.KuduScanTokenBuilder tokenBuilder = syncClient.newScanTokenBuilder(table);
+ tokenBuilder.setProjectedColumnIndexes(ImmutableList.<Integer>of());
+ List<KuduScanToken> tokens = tokenBuilder.build();
+ assertEquals(16, tokens.size());
+
+ for (KuduScanToken token : tokens) {
+ // Sanity check to make sure the debug printing does not throw.
+ LOG.debug(KuduScanToken.stringifySerializedToken(token.serialize(), syncClient));
+ }
+ } finally {
+ AsyncKuduClient.FETCH_TABLETS_PER_RANGE_LOOKUP = saveFetchTablets;
}
}