You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by jd...@apache.org on 2016/07/25 17:15:16 UTC
[11/36] incubator-kudu git commit: [java-client] repackage to
org.apache.kudu (Part 1)
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/client/TestColumnRangePredicate.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestColumnRangePredicate.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestColumnRangePredicate.java
new file mode 100644
index 0000000..bf13f2d
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestColumnRangePredicate.java
@@ -0,0 +1,72 @@
+// 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.
+package org.kududb.client;
+
+import com.google.common.collect.Lists;
+import org.junit.Test;
+import org.kududb.ColumnSchema;
+import org.kududb.Type;
+import org.kududb.tserver.Tserver;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.junit.Assert.*;
+
+public class TestColumnRangePredicate {
+
+ @Test
+ public void testRawLists() {
+ ColumnSchema col1 = new ColumnSchema.ColumnSchemaBuilder("col1", Type.INT32).build();
+ ColumnSchema col2 = new ColumnSchema.ColumnSchemaBuilder("col2", Type.STRING).build();
+
+ ColumnRangePredicate pred1 = new ColumnRangePredicate(col1);
+ pred1.setLowerBound(1);
+
+ ColumnRangePredicate pred2 = new ColumnRangePredicate(col1);
+ pred2.setUpperBound(2);
+
+ ColumnRangePredicate pred3 = new ColumnRangePredicate(col2);
+ pred3.setLowerBound("aaa");
+ pred3.setUpperBound("bbb");
+
+ List<ColumnRangePredicate> preds = Lists.newArrayList(pred1, pred2, pred3);
+
+ byte[] rawPreds = ColumnRangePredicate.toByteArray(preds);
+
+ List<Tserver.ColumnRangePredicatePB> decodedPreds = null;
+ try {
+ decodedPreds = ColumnRangePredicate.fromByteArray(rawPreds);
+ } catch (IllegalArgumentException e) {
+ fail("Couldn't decode: " + e.getMessage());
+ }
+
+ assertEquals(3, decodedPreds.size());
+
+ assertEquals(col1.getName(), decodedPreds.get(0).getColumn().getName());
+ assertEquals(1, Bytes.getInt(Bytes.get(decodedPreds.get(0).getLowerBound())));
+ assertFalse(decodedPreds.get(0).hasInclusiveUpperBound());
+
+ assertEquals(col1.getName(), decodedPreds.get(1).getColumn().getName());
+ assertEquals(2, Bytes.getInt(Bytes.get(decodedPreds.get(1).getInclusiveUpperBound())));
+ assertFalse(decodedPreds.get(1).hasLowerBound());
+
+ assertEquals(col2.getName(), decodedPreds.get(2).getColumn().getName());
+ assertEquals("aaa", Bytes.getString(Bytes.get(decodedPreds.get(2).getLowerBound())));
+ assertEquals("bbb", Bytes.getString(Bytes.get(decodedPreds.get(2).getInclusiveUpperBound())));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/client/TestDeadlineTracker.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestDeadlineTracker.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestDeadlineTracker.java
new file mode 100644
index 0000000..20ee06a
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestDeadlineTracker.java
@@ -0,0 +1,74 @@
+// 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.
+package org.kududb.client;
+
+import static org.junit.Assert.*;
+
+import com.google.common.base.Stopwatch;
+import com.google.common.base.Ticker;
+import org.junit.Test;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+public class TestDeadlineTracker {
+
+ @Test
+ public void testTimeout() {
+ final AtomicLong timeToReturn = new AtomicLong();
+ Ticker ticker = new Ticker() {
+ @Override
+ public long read() {
+ return timeToReturn.get();
+ }
+ };
+ Stopwatch stopwatch = Stopwatch.createUnstarted(ticker);
+
+ // no timeout set
+ DeadlineTracker tracker = new DeadlineTracker(stopwatch);
+ tracker.setDeadline(0);
+ assertFalse(tracker.hasDeadline());
+ assertFalse(tracker.timedOut());
+
+ // 500ms timeout set
+ tracker.reset();
+ tracker.setDeadline(500);
+ assertTrue(tracker.hasDeadline());
+ assertFalse(tracker.timedOut());
+ assertFalse(tracker.wouldSleepingTimeout(499));
+ assertTrue(tracker.wouldSleepingTimeout(500));
+ assertTrue(tracker.wouldSleepingTimeout(501));
+ assertEquals(500, tracker.getMillisBeforeDeadline());
+
+ // fast forward 200ms
+ timeToReturn.set(200 * 1000000);
+ assertTrue(tracker.hasDeadline());
+ assertFalse(tracker.timedOut());
+ assertFalse(tracker.wouldSleepingTimeout(299));
+ assertTrue(tracker.wouldSleepingTimeout(300));
+ assertTrue(tracker.wouldSleepingTimeout(301));
+ assertEquals(300, tracker.getMillisBeforeDeadline());
+
+ // fast forward another 400ms, so the RPC timed out
+ timeToReturn.set(600 * 1000000);
+ assertTrue(tracker.hasDeadline());
+ assertTrue(tracker.timedOut());
+ assertTrue(tracker.wouldSleepingTimeout(299));
+ assertTrue(tracker.wouldSleepingTimeout(300));
+ assertTrue(tracker.wouldSleepingTimeout(301));
+ assertEquals(1, tracker.getMillisBeforeDeadline());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/client/TestErrorCollector.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestErrorCollector.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestErrorCollector.java
new file mode 100644
index 0000000..883b2e1
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestErrorCollector.java
@@ -0,0 +1,90 @@
+// 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.
+package org.kududb.client;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestErrorCollector {
+
+ @Test
+ public void testErrorCollector() {
+ int maxErrors = 10;
+ ErrorCollector collector = new ErrorCollector(maxErrors);
+
+ // Test with no errors.
+ int countToTest = 0;
+ Assert.assertEquals(countToTest, collector.countErrors());
+ RowErrorsAndOverflowStatus reos = collector.getErrors();
+ Assert.assertEquals(0, collector.countErrors());
+ Assert.assertFalse(reos.isOverflowed());
+ Assert.assertEquals(countToTest, reos.getRowErrors().length);
+
+ // Test a single row error.
+ countToTest = 1;
+ collector.addError(createRowError(countToTest));
+ Assert.assertEquals(countToTest, collector.countErrors());
+ reos = collector.getErrors();
+ Assert.assertEquals(0, collector.countErrors());
+ Assert.assertFalse(reos.isOverflowed());
+ Assert.assertEquals(countToTest, reos.getRowErrors().length);
+ Assert.assertEquals(countToTest, reos.getRowErrors()[0].getErrorStatus().getPosixCode());
+
+ // Test filling the collector to the max.
+ countToTest = maxErrors;
+ fillCollectorWith(collector, countToTest);
+ Assert.assertEquals(countToTest, collector.countErrors());
+ reos = collector.getErrors();
+ Assert.assertEquals(0, collector.countErrors());
+ Assert.assertFalse(reos.isOverflowed());
+ Assert.assertEquals(countToTest, reos.getRowErrors().length);
+ Assert.assertEquals(countToTest - 1, reos.getRowErrors()[9].getErrorStatus().getPosixCode());
+
+ // Test overflowing.
+ countToTest = 95;
+ fillCollectorWith(collector, countToTest);
+ Assert.assertEquals(maxErrors, collector.countErrors());
+ reos = collector.getErrors();
+ Assert.assertEquals(0, collector.countErrors());
+ Assert.assertTrue(reos.isOverflowed());
+ Assert.assertEquals(maxErrors, reos.getRowErrors().length);
+ Assert.assertEquals(countToTest - 1, reos.getRowErrors()[9].getErrorStatus().getPosixCode());
+
+ // Test overflowing on a newly created collector.
+ countToTest = 95;
+ collector = new ErrorCollector(maxErrors);
+ fillCollectorWith(collector, countToTest);
+ Assert.assertEquals(maxErrors, collector.countErrors());
+ reos = collector.getErrors();
+ Assert.assertEquals(0, collector.countErrors());
+ Assert.assertTrue(reos.isOverflowed());
+ Assert.assertEquals(maxErrors, reos.getRowErrors().length);
+ Assert.assertEquals(countToTest - 1, reos.getRowErrors()[9].getErrorStatus().getPosixCode());
+ }
+
+ private void fillCollectorWith(ErrorCollector collector, int errorsToAdd) {
+ for (int i = 0; i < errorsToAdd; i++) {
+ collector.addError(createRowError(i));
+ }
+ }
+
+ private RowError createRowError(int id) {
+ // Use the error status as a way to message pass and so that we can test we're getting the right
+ // messages on the other end.
+ return new RowError(Status.NotAuthorized("test", id), null, "test");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/client/TestFlexiblePartitioning.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestFlexiblePartitioning.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestFlexiblePartitioning.java
new file mode 100644
index 0000000..dafd74a
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestFlexiblePartitioning.java
@@ -0,0 +1,422 @@
+// 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.
+package org.kududb.client;
+
+import com.google.common.base.Predicate;
+import com.google.common.base.Predicates;
+import com.google.common.collect.ComparisonChain;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Sets;
+import org.junit.Before;
+import org.junit.Test;
+import org.kududb.ColumnSchema;
+import org.kududb.Schema;
+import org.kududb.Type;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+public class TestFlexiblePartitioning extends BaseKuduTest {
+ private String tableName;
+
+ @Before
+ public void setTableName() {
+ tableName = TestKuduClient.class.getName() + "-" + System.currentTimeMillis();
+ }
+
+ private static Schema createSchema() {
+ ArrayList<ColumnSchema> columns = new ArrayList<>(3);
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("a", Type.STRING).key(true).build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("b", Type.STRING).key(true).build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("c", Type.STRING).key(true).build());
+ return new Schema(columns);
+ }
+
+ private static Set<Row> rows() throws Exception {
+ Set<Row> rows = new HashSet<>();
+ for (int a = 0; a < 6; a++) {
+ for (int b = 0; b < 6; b++) {
+ for (int c = 0; c < 6; c++) {
+ rows.add(new Row(String.format("%s", a),
+ String.format("%s", b),
+ String.format("%s", c)));
+ }
+ }
+ }
+ return rows;
+ }
+
+ private void insertRows(KuduTable table, Set<Row> rows) throws Exception {
+ KuduSession session = syncClient.newSession();
+ try {
+ for (Row row : rows) {
+ Insert insert = table.newInsert();
+ PartialRow insertRow = insert.getRow();
+ row.fillPartialRow(insertRow);
+ session.apply(insert);
+ }
+ } finally {
+ session.close();
+ }
+ }
+
+ private Set<Row> collectRows(KuduScanner scanner) throws Exception {
+ Set<Row> rows = new HashSet<>();
+ while (scanner.hasMoreRows()) {
+ for (RowResult result : scanner.nextRows()) {
+ rows.add(Row.fromResult(result));
+ }
+ }
+ return rows;
+ }
+
+ private void testPartitionSchema(CreateTableOptions tableBuilder) throws Exception {
+ Schema schema = createSchema();
+
+ syncClient.createTable(tableName, schema, tableBuilder);
+
+ KuduTable table = syncClient.openTable(tableName);
+
+ Set<Row> rows = rows();
+ insertRows(table, rows);
+
+ // Full table scan
+ assertEquals(rows, collectRows(syncClient.newScannerBuilder(table).build()));
+
+ { // Lower bound
+ Row minRow = new Row("1", "3", "5");
+ PartialRow lowerBound = schema.newPartialRow();
+ minRow.fillPartialRow(lowerBound);
+
+ Set<Row> expected = Sets.filter(rows, minRow.gtePred());
+
+ KuduScanner scanner = syncClient.newScannerBuilder(table).lowerBound(lowerBound).build();
+ Set<Row> results = collectRows(scanner);
+
+ assertEquals(expected, results);
+ }
+
+ { // Upper bound
+ Row maxRow = new Row("1", "3", "5");
+ PartialRow upperBound = schema.newPartialRow();
+ maxRow.fillPartialRow(upperBound);
+
+ Set<Row> expected = Sets.filter(rows, maxRow.ltPred());
+
+ KuduScanner scanner = syncClient.newScannerBuilder(table)
+ .exclusiveUpperBound(upperBound)
+ .build();
+ Set<Row> results = collectRows(scanner);
+
+ assertEquals(expected, results);
+ }
+
+ { // Lower & Upper bounds
+ Row minRow = new Row("1", "3", "5");
+ Row maxRow = new Row("2", "4", "");
+ PartialRow lowerBound = schema.newPartialRow();
+ minRow.fillPartialRow(lowerBound);
+ PartialRow upperBound = schema.newPartialRow();
+ maxRow.fillPartialRow(upperBound);
+
+ Set<Row> expected = Sets.filter(rows, Predicates.and(minRow.gtePred(), maxRow.ltPred()));
+
+ KuduScanner scanner = syncClient.newScannerBuilder(table)
+ .lowerBound(lowerBound)
+ .exclusiveUpperBound(upperBound)
+ .build();
+ Set<Row> results = collectRows(scanner);
+
+ assertEquals(expected, results);
+ }
+
+ List<LocatedTablet> tablets = table.getTabletsLocations(TestTimeouts.DEFAULT_SLEEP);
+
+ { // Per-tablet scan
+ Set<Row> results = new HashSet<>();
+
+ for (LocatedTablet tablet : tablets) {
+ KuduScanner scanner = syncClient.newScannerBuilder(table)
+ .lowerBoundPartitionKeyRaw(tablet.getPartition().getPartitionKeyStart())
+ .exclusiveUpperBoundPartitionKeyRaw(tablet.getPartition().getPartitionKeyEnd())
+ .build();
+ Set<Row> tabletResults = collectRows(scanner);
+ Set<Row> intersection = Sets.intersection(results, tabletResults);
+ assertEquals(new HashSet<>(), intersection);
+ results.addAll(tabletResults);
+ }
+
+ assertEquals(rows, results);
+ }
+
+ { // Per-tablet scan with lower & upper bounds
+ Row minRow = new Row("1", "3", "5");
+ Row maxRow = new Row("2", "4", "");
+ PartialRow lowerBound = schema.newPartialRow();
+ minRow.fillPartialRow(lowerBound);
+ PartialRow upperBound = schema.newPartialRow();
+ maxRow.fillPartialRow(upperBound);
+
+ Set<Row> expected = Sets.filter(rows, Predicates.and(minRow.gtePred(), maxRow.ltPred()));
+ Set<Row> results = new HashSet<>();
+
+ for (LocatedTablet tablet : tablets) {
+ KuduScanner scanner = syncClient.newScannerBuilder(table)
+ .lowerBound(lowerBound)
+ .exclusiveUpperBound(upperBound)
+ .lowerBoundPartitionKeyRaw(tablet.getPartition().getPartitionKeyStart())
+ .exclusiveUpperBoundPartitionKeyRaw(tablet.getPartition().getPartitionKeyEnd())
+ .build();
+ Set<Row> tabletResults = collectRows(scanner);
+ Set<Row> intersection = Sets.intersection(results, tabletResults);
+ assertEquals(new HashSet<>(), intersection);
+ results.addAll(tabletResults);
+ }
+
+ assertEquals(expected, results);
+ }
+ }
+
+ @Test(timeout = 100000)
+ public void testHashBucketedTable() throws Exception {
+ CreateTableOptions tableBuilder = new CreateTableOptions();
+ tableBuilder.addHashPartitions(ImmutableList.of("a"), 3);
+ tableBuilder.addHashPartitions(ImmutableList.of("b", "c"), 3, 42);
+ tableBuilder.setRangePartitionColumns(ImmutableList.<String>of());
+ testPartitionSchema(tableBuilder);
+ }
+
+ @Test(timeout = 100000)
+ public void testNonDefaultRangePartitionedTable() throws Exception {
+ Schema schema = createSchema();
+ CreateTableOptions tableBuilder = new CreateTableOptions();
+ tableBuilder.setRangePartitionColumns(ImmutableList.of("c", "b"));
+
+ PartialRow split = schema.newPartialRow();
+ split.addString("c", "3");
+ tableBuilder.addSplitRow(split);
+
+ split = schema.newPartialRow();
+ split.addString("c", "3");
+ split.addString("b", "3");
+ tableBuilder.addSplitRow(split);
+
+ testPartitionSchema(tableBuilder);
+ }
+
+ @Test(timeout = 100000)
+ public void testHashBucketedAndRangePartitionedTable() throws Exception {
+ Schema schema = createSchema();
+ CreateTableOptions tableBuilder = new CreateTableOptions();
+ tableBuilder.addHashPartitions(ImmutableList.of("a"), 3);
+ tableBuilder.addHashPartitions(ImmutableList.of("b", "c"), 3, 42);
+ tableBuilder.setRangePartitionColumns(ImmutableList.of("c", "b"));
+
+ PartialRow split = schema.newPartialRow();
+ split.addString("c", "3");
+ tableBuilder.addSplitRow(split);
+
+ split = schema.newPartialRow();
+ split.addString("c", "3");
+ split.addString("b", "3");
+ tableBuilder.addSplitRow(split);
+
+ testPartitionSchema(tableBuilder);
+ }
+
+ @Test(timeout = 100000)
+ public void testNonCoveredRangePartitionedTable() throws Exception {
+ Schema schema = createSchema();
+ CreateTableOptions tableBuilder = new CreateTableOptions();
+ tableBuilder.setRangePartitionColumns(ImmutableList.of("a", "b", "c"));
+
+ // Create a non covered range between (3, 5, 6) and (4, 0, 0)
+
+ PartialRow lowerBoundA = schema.newPartialRow();
+ lowerBoundA.addString("a", "0");
+ lowerBoundA.addString("b", "0");
+ lowerBoundA.addString("c", "0");
+ PartialRow upperBoundA = schema.newPartialRow();
+ upperBoundA.addString("a", "3");
+ upperBoundA.addString("b", "5");
+ upperBoundA.addString("b", "6");
+ tableBuilder.addRangeBound(lowerBoundA, upperBoundA);
+
+ PartialRow lowerBoundB = schema.newPartialRow();
+ lowerBoundB.addString("a", "4");
+ lowerBoundB.addString("b", "0");
+ lowerBoundB.addString("c", "0");
+ PartialRow upperBoundB = schema.newPartialRow();
+ upperBoundB.addString("a", "5");
+ upperBoundB.addString("b", "5");
+ upperBoundB.addString("b", "6");
+ tableBuilder.addRangeBound(lowerBoundB, upperBoundB);
+
+ testPartitionSchema(tableBuilder);
+ }
+
+ @Test(timeout = 100000)
+ public void testHashBucketedAndNonCoveredRangePartitionedTable() throws Exception {
+ Schema schema = createSchema();
+ CreateTableOptions tableBuilder = new CreateTableOptions();
+ tableBuilder.setRangePartitionColumns(ImmutableList.of("a", "b", "c"));
+
+ // Create a non covered range between (3, 5, 6) and (4, 0, 0)
+
+ PartialRow lowerBoundA = schema.newPartialRow();
+ lowerBoundA.addString("a", "0");
+ lowerBoundA.addString("b", "0");
+ lowerBoundA.addString("c", "0");
+ PartialRow upperBoundA = schema.newPartialRow();
+ upperBoundA.addString("a", "3");
+ upperBoundA.addString("b", "5");
+ upperBoundA.addString("c", "6");
+ tableBuilder.addRangeBound(lowerBoundA, upperBoundA);
+
+ PartialRow lowerBoundB = schema.newPartialRow();
+ lowerBoundB.addString("a", "4");
+ lowerBoundB.addString("b", "0");
+ lowerBoundB.addString("c", "0");
+ PartialRow upperBoundB = schema.newPartialRow();
+ upperBoundB.addString("a", "5");
+ upperBoundB.addString("b", "5");
+ upperBoundB.addString("c", "6");
+ tableBuilder.addRangeBound(lowerBoundB, upperBoundB);
+
+ tableBuilder.addHashPartitions(ImmutableList.of("a", "b", "c"), 4);
+
+ testPartitionSchema(tableBuilder);
+ }
+
+ @Test(timeout = 100000)
+ public void testSimplePartitionedTable() throws Exception {
+ Schema schema = createSchema();
+ CreateTableOptions tableBuilder =
+ new CreateTableOptions().setRangePartitionColumns(ImmutableList.of("a", "b", "c"));
+
+ PartialRow split = schema.newPartialRow();
+ split.addString("c", "3");
+ tableBuilder.addSplitRow(split);
+
+ split = schema.newPartialRow();
+ split.addString("c", "3");
+ split.addString("b", "3");
+ tableBuilder.addSplitRow(split);
+
+ testPartitionSchema(tableBuilder);
+ }
+
+ @Test(timeout = 100000)
+ public void testUnpartitionedTable() throws Exception {
+ CreateTableOptions tableBuilder =
+ new CreateTableOptions().setRangePartitionColumns(ImmutableList.<String>of());
+ testPartitionSchema(tableBuilder);
+ }
+
+ public static class Row implements Comparable<Row> {
+ private final String a;
+ private final String b;
+ private final String c;
+
+ public Row(String a, String b, String c) {
+ this.a = a;
+ this.b = b;
+ this.c = c;
+ }
+
+ public String getA() {
+ return a;
+ }
+
+ public String getB() {
+ return b;
+ }
+
+ public String getC() {
+ return c;
+ }
+
+ public void fillPartialRow(PartialRow row) {
+ row.addString("a", a);
+ row.addString("b", b);
+ row.addString("c", c);
+ }
+
+ private static Row fromResult(RowResult result) {
+ return new Row(result.getString("a"),
+ result.getString("b"),
+ result.getString("c"));
+ }
+
+ public Predicate<Row> gtePred() {
+ return new Predicate<Row>() {
+ @Override
+ public boolean apply(Row other) {
+ return other.compareTo(Row.this) >= 0;
+ }
+ };
+ }
+
+ public Predicate<Row> ltPred() {
+ return new Predicate<Row>() {
+ @Override
+ public boolean apply(Row other) {
+ return other.compareTo(Row.this) < 0;
+ }
+ };
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ Row row = (Row) o;
+ return Objects.equals(a, row.a)
+ && Objects.equals(b, row.b)
+ && Objects.equals(c, row.c);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(a, b, c);
+ }
+
+ @Override
+ public int compareTo(Row other) {
+ return ComparisonChain.start()
+ .compare(a, other.a)
+ .compare(b, other.b)
+ .compare(c, other.c)
+ .result();
+ }
+
+ @Override
+ public String toString() {
+ return com.google.common.base.Objects.toStringHelper(this)
+ .add("a", a)
+ .add("b", b)
+ .add("c", c)
+ .toString();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/client/TestHybridTime.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestHybridTime.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestHybridTime.java
new file mode 100644
index 0000000..666ac38
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestHybridTime.java
@@ -0,0 +1,163 @@
+// 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.
+package org.kududb.client;
+
+import com.google.common.collect.ImmutableList;
+import com.stumbleupon.async.Deferred;
+import org.kududb.ColumnSchema;
+import org.kududb.Schema;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static org.kududb.Type.STRING;
+import static org.kududb.client.ExternalConsistencyMode.CLIENT_PROPAGATED;
+import static org.kududb.util.HybridTimeUtil.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * This only tests client propagation since it's the only thing that is client-specific.
+ * All the work for commit wait is done and tested on the server-side.
+ */
+public class TestHybridTime extends BaseKuduTest {
+ private static final Logger LOG = LoggerFactory.getLogger(TestHybridTime.class);
+
+ // Generate a unique table name
+ protected static final String TABLE_NAME =
+ TestHybridTime.class.getName() + "-" + System.currentTimeMillis();
+
+ protected static Schema schema = getSchema();
+ protected static KuduTable table;
+
+ @BeforeClass
+ public static void setUpBeforeClass() throws Exception {
+ BaseKuduTest.setUpBeforeClass();
+
+ // Using multiple tablets doesn't work with the current way this test works since we could
+ // jump from one TS to another which changes the logical clock.
+ CreateTableOptions builder =
+ new CreateTableOptions().setRangePartitionColumns(ImmutableList.of("key"));
+ table = createTable(TABLE_NAME, schema, builder);
+ }
+
+ private static Schema getSchema() {
+ ArrayList<ColumnSchema> columns = new ArrayList<ColumnSchema>(1);
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("key", STRING)
+ .key(true)
+ .build());
+ return new Schema(columns);
+ }
+
+ /**
+ * We write three rows. We increment the timestamp we get back from the first write
+ * by some amount. The remaining writes should force an update to the server's clock and
+ * only increment the logical values.
+ *
+ * @throws Exception
+ */
+ @Test(timeout = 100000)
+ public void test() throws Exception {
+ AsyncKuduSession session = client.newSession();
+ session.setFlushMode(AsyncKuduSession.FlushMode.AUTO_FLUSH_SYNC);
+ session.setExternalConsistencyMode(CLIENT_PROPAGATED);
+ long[] clockValues;
+ long previousLogicalValue = 0;
+ long previousPhysicalValue = 0;
+
+ // Test timestamp propagation with single operations
+ String[] keys = new String[] {"1", "2", "3"};
+ for (int i = 0; i < keys.length; i++) {
+ Insert insert = table.newInsert();
+ PartialRow row = insert.getRow();
+ row.addString(schema.getColumnByIndex(0).getName(), keys[i]);
+ Deferred<OperationResponse> d = session.apply(insert);
+ OperationResponse response = d.join(DEFAULT_SLEEP);
+ assertTrue(response.getWriteTimestamp() != 0);
+ clockValues = HTTimestampToPhysicalAndLogical(response.getWriteTimestamp());
+ LOG.debug("Clock value after write[" + i + "]: " + new Date(clockValues[0] / 1000).toString()
+ + " Logical value: " + clockValues[1]);
+ // on the very first write we update the clock into the future
+ // so that remaining writes only update logical values
+ if (i == 0) {
+ assertEquals(clockValues[1], 0);
+ long toUpdateTs = clockValues[0] + 5000000;
+ previousPhysicalValue = toUpdateTs;
+ // After the first write we fake-update the clock into the future. Following writes
+ // should force the servers to update their clocks to this value.
+ client.updateLastPropagatedTimestamp(
+ clockTimestampToHTTimestamp(toUpdateTs, TimeUnit.MICROSECONDS));
+ } else {
+ assertEquals(clockValues[0], previousPhysicalValue);
+ assertTrue(clockValues[1] > previousLogicalValue);
+ previousLogicalValue = clockValues[1];
+ }
+ }
+
+ // Test timestamp propagation with Batches
+ session.setFlushMode(AsyncKuduSession.FlushMode.MANUAL_FLUSH);
+ keys = new String[] {"11", "22", "33"};
+ for (int i = 0; i < keys.length; i++) {
+ Insert insert = table.newInsert();
+ PartialRow row = insert.getRow();
+ row.addString(schema.getColumnByIndex(0).getName(), keys[i]);
+ session.apply(insert);
+ Deferred<List<OperationResponse>> d = session.flush();
+ List<OperationResponse> responses = d.join(DEFAULT_SLEEP);
+ assertEquals("Response was not of the expected size: " + responses.size(),
+ 1, responses.size());
+
+ OperationResponse response = responses.get(0);
+ assertTrue(response.getWriteTimestamp() != 0);
+ clockValues = HTTimestampToPhysicalAndLogical(response.getWriteTimestamp());
+ LOG.debug("Clock value after write[" + i + "]: " + new Date(clockValues[0] / 1000).toString()
+ + " Logical value: " + clockValues[1]);
+ assertEquals(clockValues[0], previousPhysicalValue);
+ assertTrue(clockValues[1] > previousLogicalValue);
+ previousLogicalValue = clockValues[1];
+ }
+
+ // Scan all rows with READ_LATEST (the default) we should get 6 rows back
+ assertEquals(6, countRowsInScan(client.newScannerBuilder(table).build()));
+
+ // Now scan at multiple instances with READ_AT_SNAPSHOT we should get different
+ // counts depending on the scan timestamp.
+ long snapTime = physicalAndLogicalToHTTimestamp(previousPhysicalValue, 0);
+ assertEquals(1, scanAtSnapshot(snapTime));
+ snapTime = physicalAndLogicalToHTTimestamp(previousPhysicalValue, 5);
+ assertEquals(4, scanAtSnapshot(snapTime));
+ // Our last snap time needs to one one into the future w.r.t. the last write's timestamp
+ // for us to be able to get all rows, but the snap timestamp can't be bigger than the prop.
+ // timestamp so we increase both.
+ client.updateLastPropagatedTimestamp(client.getLastPropagatedTimestamp() + 1);
+ snapTime = physicalAndLogicalToHTTimestamp(previousPhysicalValue, previousLogicalValue + 1);
+ assertEquals(6, scanAtSnapshot(snapTime));
+ }
+
+ private int scanAtSnapshot(long time) throws Exception {
+ AsyncKuduScanner.AsyncKuduScannerBuilder builder = client.newScannerBuilder(table)
+ .snapshotTimestampRaw(time)
+ .readMode(AsyncKuduScanner.ReadMode.READ_AT_SNAPSHOT);
+ return countRowsInScan(builder.build());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/client/TestKeyEncoding.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKeyEncoding.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKeyEncoding.java
new file mode 100644
index 0000000..e446445
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKeyEncoding.java
@@ -0,0 +1,272 @@
+// 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.
+package org.kududb.client;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import com.google.common.base.Charsets;
+import com.google.common.collect.ImmutableList;
+import org.junit.Test;
+import org.kududb.ColumnSchema;
+import org.kududb.ColumnSchema.ColumnSchemaBuilder;
+import org.kududb.Common;
+import org.kududb.Schema;
+import org.kududb.Type;
+import org.kududb.client.PartitionSchema.HashBucketSchema;
+import org.kududb.client.PartitionSchema.RangeSchema;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class TestKeyEncoding {
+
+ private static Schema buildSchema(ColumnSchemaBuilder... columns) {
+ int i = 0;
+ Common.SchemaPB.Builder pb = Common.SchemaPB.newBuilder();
+ for (ColumnSchemaBuilder column : columns) {
+ Common.ColumnSchemaPB.Builder columnPb =
+ ProtobufHelper.columnToPb(column.build()).toBuilder();
+ columnPb.setId(i++);
+ pb.addColumns(columnPb);
+ }
+ return ProtobufHelper.pbToSchema(pb.build());
+ }
+
+ private static void assertBytesEquals(byte[] actual, byte[] expected) {
+ assertTrue(String.format("expected: '%s', got '%s'",
+ Bytes.pretty(expected),
+ Bytes.pretty(actual)),
+ Bytes.equals(expected, actual));
+ }
+
+ private static void assertBytesEquals(byte[] actual, String expected) {
+ assertBytesEquals(actual, expected.getBytes(Charsets.UTF_8));
+ }
+
+ /**
+ * Builds the default partition schema for a schema.
+ * @param schema the schema
+ * @return a default partition schema
+ */
+ private PartitionSchema defaultPartitionSchema(Schema schema) {
+ List<Integer> columnIds = new ArrayList<>();
+ for (int i = 0; i < schema.getPrimaryKeyColumnCount(); i++) {
+ // Schema does not provide a way to lookup a column ID by column index,
+ // so instead we assume that the IDs for the primary key columns match
+ // their respective index, which holds up when the schema is created
+ // with buildSchema.
+ columnIds.add(i);
+ }
+ return new PartitionSchema(
+ new PartitionSchema.RangeSchema(columnIds),
+ ImmutableList.<PartitionSchema.HashBucketSchema>of(), schema);
+ }
+
+ @Test
+ public void testPrimaryKeys() {
+ Schema schemaOneString =
+ buildSchema(new ColumnSchema.ColumnSchemaBuilder("key", Type.STRING).key(true));
+ KuduTable table = new KuduTable(null, "one", "one", schemaOneString,
+ defaultPartitionSchema(schemaOneString));
+ Insert oneKeyInsert = new Insert(table);
+ PartialRow row = oneKeyInsert.getRow();
+ row.addString("key", "foo");
+ assertBytesEquals(row.encodePrimaryKey(), "foo");
+
+ Schema schemaTwoString = buildSchema(
+ new ColumnSchema.ColumnSchemaBuilder("key", Type.STRING).key(true),
+ new ColumnSchema.ColumnSchemaBuilder("key2", Type.STRING).key(true));
+ KuduTable table2 = new KuduTable(null, "two", "two", schemaTwoString,
+ defaultPartitionSchema(schemaTwoString));
+ Insert twoKeyInsert = new Insert(table2);
+ row = twoKeyInsert.getRow();
+ row.addString("key", "foo");
+ row.addString("key2", "bar");
+ assertBytesEquals(row.encodePrimaryKey(), "foo\0\0bar");
+
+ Insert twoKeyInsertWithNull = new Insert(table2);
+ row = twoKeyInsertWithNull.getRow();
+ row.addString("key", "xxx\0yyy");
+ row.addString("key2", "bar");
+ assertBytesEquals(row.encodePrimaryKey(), "xxx\0\1yyy\0\0bar");
+
+ // test that we get the correct memcmp result, the bytes are in big-endian order in a key
+ Schema schemaIntString = buildSchema(
+ new ColumnSchema.ColumnSchemaBuilder("key", Type.INT32).key(true),
+ new ColumnSchema.ColumnSchemaBuilder("key2", Type.STRING).key(true));
+ PartitionSchema partitionSchemaIntString = defaultPartitionSchema(schemaIntString);
+ KuduTable table3 = new KuduTable(null, "three", "three",
+ schemaIntString, partitionSchemaIntString);
+ Insert small = new Insert(table3);
+ row = small.getRow();
+ row.addInt("key", 20);
+ row.addString("key2", "data");
+ byte[] smallPK = small.getRow().encodePrimaryKey();
+ assertEquals(0, Bytes.memcmp(smallPK, smallPK));
+
+ Insert big = new Insert(table3);
+ row = big.getRow();
+ row.addInt("key", 10000);
+ row.addString("key2", "data");
+ byte[] bigPK = big.getRow().encodePrimaryKey();
+ assertTrue(Bytes.memcmp(smallPK, bigPK) < 0);
+ assertTrue(Bytes.memcmp(bigPK, smallPK) > 0);
+
+ // The following tests test our assumptions on unsigned data types sorting from KeyEncoder
+ byte four = 4;
+ byte onHundredTwentyFour = -4;
+ four = Bytes.xorLeftMostBit(four);
+ onHundredTwentyFour = Bytes.xorLeftMostBit(onHundredTwentyFour);
+ assertTrue(four < onHundredTwentyFour);
+
+ byte[] threeHundred = Bytes.fromInt(300);
+ byte[] reallyBigNumber = Bytes.fromInt(-300);
+ threeHundred[0] = Bytes.xorLeftMostBit(threeHundred[0]);
+ reallyBigNumber[3] = Bytes.xorLeftMostBit(reallyBigNumber[3]);
+ assertTrue(Bytes.memcmp(threeHundred, reallyBigNumber) < 0);
+ }
+
+ @Test
+ public void testPrimaryKeyEncoding() {
+ Schema schema = buildSchema(
+ new ColumnSchemaBuilder("int8", Type.INT8).key(true),
+ new ColumnSchemaBuilder("int16", Type.INT16).key(true),
+ new ColumnSchemaBuilder("int32", Type.INT32).key(true),
+ new ColumnSchemaBuilder("int64", Type.INT64).key(true),
+ new ColumnSchemaBuilder("string", Type.STRING).key(true),
+ new ColumnSchemaBuilder("binary", Type.BINARY).key(true));
+
+ PartialRow rowA = schema.newPartialRow();
+ rowA.addByte("int8", Byte.MIN_VALUE);
+ rowA.addShort("int16", Short.MIN_VALUE);
+ rowA.addInt("int32", Integer.MIN_VALUE);
+ rowA.addLong("int64", Long.MIN_VALUE);
+ rowA.addString("string", "");
+ rowA.addBinary("binary", "".getBytes(Charsets.UTF_8));
+
+ assertBytesEquals(rowA.encodePrimaryKey(),
+ "\0"
+ + "\0\0"
+ + "\0\0\0\0"
+ + "\0\0\0\0\0\0\0\0"
+ + "\0\0"
+ + "");
+
+ PartialRow rowB = schema.newPartialRow();
+ rowB.addByte("int8", Byte.MAX_VALUE);
+ rowB.addShort("int16", Short.MAX_VALUE);
+ rowB.addInt("int32", Integer.MAX_VALUE);
+ rowB.addLong("int64", Long.MAX_VALUE);
+ rowB.addString("string", "abc\1\0def");
+ rowB.addBinary("binary", "\0\1binary".getBytes(Charsets.UTF_8));
+
+ assertBytesEquals(rowB.encodePrimaryKey(),
+ new byte[] {
+ -1,
+ -1, -1,
+ -1, -1, -1, -1,
+ -1, -1, -1, -1, -1, -1, -1, -1,
+ 'a', 'b', 'c', 1, 0, 1, 'd', 'e', 'f', 0, 0,
+ 0, 1, 'b', 'i', 'n', 'a', 'r', 'y',
+ });
+
+ PartialRow rowC = schema.newPartialRow();
+ rowC.addByte("int8", (byte) 1);
+ rowC.addShort("int16", (short) 2);
+ rowC.addInt("int32", 3);
+ rowC.addLong("int64", 4);
+ rowC.addString("string", "abc\n123");
+ rowC.addBinary("binary", "\0\1\2\3\4\5".getBytes(Charsets.UTF_8));
+
+ assertBytesEquals(rowC.encodePrimaryKey(),
+ new byte[] {
+ (byte) 0x81,
+ (byte) 0x80, 2,
+ (byte) 0x80, 0, 0, 3,
+ (byte) 0x80, 0, 0, 0, 0, 0, 0, 4,
+ 'a', 'b', 'c', '\n', '1', '2', '3', 0, 0,
+ 0, 1, 2, 3, 4, 5,
+ });
+ }
+
+ @Test
+ public void testPartitionKeyEncoding() {
+ KeyEncoder encoder = new KeyEncoder();
+ Schema schema = buildSchema(
+ new ColumnSchemaBuilder("a", Type.INT32).key(true),
+ new ColumnSchemaBuilder("b", Type.STRING).key(true),
+ new ColumnSchemaBuilder("c", Type.STRING).key(true));
+
+ PartitionSchema partitionSchema =
+ new PartitionSchema(new RangeSchema(ImmutableList.of(0, 1, 2)),
+ ImmutableList.of(
+ new HashBucketSchema(ImmutableList.of(0, 1), 32, 0),
+ new HashBucketSchema(ImmutableList.of(2), 32, 42)),
+ schema);
+
+ PartialRow rowA = schema.newPartialRow();
+ rowA.addInt("a", 0);
+ rowA.addString("b", "");
+ rowA.addString("c", "");
+ assertBytesEquals(encoder.encodePartitionKey(rowA, partitionSchema),
+ new byte[]{
+ 0, 0, 0, 0, // hash(0, "")
+ 0, 0, 0, 0x14, // hash("")
+ (byte) 0x80, 0, 0, 0, // a = 0
+ 0, 0, // b = ""; c is elided
+ });
+
+ PartialRow rowB = schema.newPartialRow();
+ rowB.addInt("a", 1);
+ rowB.addString("b", "");
+ rowB.addString("c", "");
+ assertBytesEquals(encoder.encodePartitionKey(rowB, partitionSchema),
+ new byte[]{
+ 0, 0, 0, 0x5, // hash(1, "")
+ 0, 0, 0, 0x14, // hash("")
+ (byte) 0x80, 0, 0, 1, // a = 0
+ 0, 0, // b = ""; c is elided
+ });
+
+ PartialRow rowC = schema.newPartialRow();
+ rowC.addInt("a", 0);
+ rowC.addString("b", "b");
+ rowC.addString("c", "c");
+ assertBytesEquals(encoder.encodePartitionKey(rowC, partitionSchema),
+ new byte[]{
+ 0, 0, 0, 0x1A, // hash(0, "b")
+ 0, 0, 0, 0x1D, // hash("c")
+ (byte) 0x80, 0, 0, 0, // a = 1
+ 'b', 0, 0, // b = "b"
+ 'c' // b = "c"
+ });
+
+ PartialRow rowD = schema.newPartialRow();
+ rowD.addInt("a", 1);
+ rowD.addString("b", "b");
+ rowD.addString("c", "c");
+ assertBytesEquals(encoder.encodePartitionKey(rowD, partitionSchema),
+ new byte[]{
+ 0, 0, 0, 0, // hash(1, "b")
+ 0, 0, 0, 0x1D, // hash("c")
+ (byte) 0x80, 0, 0, 1, // a = 0
+ 'b', 0, 0, // b = "b"
+ 'c' // b = "c"
+ });
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/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
new file mode 100644
index 0000000..3591b7b
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
@@ -0,0 +1,535 @@
+// 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.
+package org.kududb.client;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.kududb.client.KuduPredicate.ComparisonOp.GREATER;
+import static org.kududb.client.KuduPredicate.ComparisonOp.GREATER_EQUAL;
+import static org.kududb.client.KuduPredicate.ComparisonOp.LESS;
+import static org.kududb.client.KuduPredicate.ComparisonOp.LESS_EQUAL;
+import static org.kududb.client.RowResult.timestampToString;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Lists;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.kududb.ColumnSchema;
+import org.kududb.Schema;
+import org.kududb.Type;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestKuduClient extends BaseKuduTest {
+ private static final Logger LOG = LoggerFactory.getLogger(TestKuduClient.class);
+ private String tableName;
+
+ @Before
+ public void setTableName() {
+ tableName = TestKuduClient.class.getName() + "-" + System.currentTimeMillis();
+ }
+
+ private Schema createManyStringsSchema() {
+ ArrayList<ColumnSchema> columns = new ArrayList<ColumnSchema>(4);
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("key", Type.STRING).key(true).build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("c1", Type.STRING).build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("c2", Type.STRING).build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("c3", Type.STRING).nullable(true).build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("c4", Type.STRING).nullable(true).build());
+ return new Schema(columns);
+ }
+
+ private Schema createSchemaWithBinaryColumns() {
+ ArrayList<ColumnSchema> columns = new ArrayList<ColumnSchema>();
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("key", Type.BINARY).key(true).build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("c1", Type.STRING).build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("c2", Type.DOUBLE).build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("c3", Type.BINARY).nullable(true).build());
+ return new Schema(columns);
+ }
+
+ private Schema createSchemaWithTimestampColumns() {
+ ArrayList<ColumnSchema> columns = new ArrayList<ColumnSchema>();
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("key", Type.TIMESTAMP).key(true).build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("c1", Type.TIMESTAMP).nullable(true).build());
+ return new Schema(columns);
+ }
+
+ private static CreateTableOptions createTableOptions() {
+ return new CreateTableOptions().setRangePartitionColumns(ImmutableList.of("key"));
+ }
+
+ /**
+ * Test creating and deleting a table through a KuduClient.
+ */
+ @Test(timeout = 100000)
+ public void testCreateDeleteTable() throws Exception {
+ // Check that we can create a table.
+ syncClient.createTable(tableName, basicSchema, getBasicCreateTableOptions());
+ assertFalse(syncClient.getTablesList().getTablesList().isEmpty());
+ assertTrue(syncClient.getTablesList().getTablesList().contains(tableName));
+
+ // Check that we can delete it.
+ syncClient.deleteTable(tableName);
+ assertFalse(syncClient.getTablesList().getTablesList().contains(tableName));
+
+ // Check that we can re-recreate it, with a different schema.
+ List<ColumnSchema> columns = new ArrayList<>(basicSchema.getColumns());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("one more", Type.STRING).build());
+ Schema newSchema = new Schema(columns);
+ syncClient.createTable(tableName, newSchema, getBasicCreateTableOptions());
+
+ // Check that we can open a table and see that it has the new schema.
+ KuduTable table = syncClient.openTable(tableName);
+ assertEquals(newSchema.getColumnCount(), table.getSchema().getColumnCount());
+ assertTrue(table.getPartitionSchema().isSimpleRangePartitioning());
+
+ // Check that the block size parameter we specified in the schema is respected.
+ assertEquals(4096, newSchema.getColumn("column3_s").getDesiredBlockSize());
+ assertEquals(ColumnSchema.Encoding.DICT_ENCODING,
+ newSchema.getColumn("column3_s").getEncoding());
+ assertEquals(ColumnSchema.CompressionAlgorithm.LZ4,
+ newSchema.getColumn("column3_s").getCompressionAlgorithm());
+ }
+
+ /**
+ * Test inserting and retrieving string columns.
+ */
+ @Test(timeout = 100000)
+ public void testStrings() throws Exception {
+ Schema schema = createManyStringsSchema();
+ syncClient.createTable(tableName, schema, createTableOptions());
+
+ KuduSession session = syncClient.newSession();
+ 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("c2", "c2_" + i);
+ if (i % 2 == 1) {
+ row.addString("c3", "c3_" + i);
+ }
+ row.addString("c4", "c4_" + i);
+ // NOTE: we purposefully add the strings in a non-left-to-right
+ // order to verify that we still place them in the right position in
+ // the row.
+ row.addString("c1", "c1_" + i);
+ session.apply(insert);
+ if (i % 50 == 0) {
+ session.flush();
+ }
+ }
+ session.flush();
+
+ List<String> rowStrings = scanTableToStrings(table);
+ assertEquals(100, rowStrings.size());
+ assertEquals(
+ "STRING key=key_03, STRING c1=c1_3, STRING c2=c2_3, STRING c3=c3_3, STRING c4=c4_3",
+ rowStrings.get(3));
+ assertEquals(
+ "STRING key=key_04, STRING c1=c1_4, STRING c2=c2_4, STRING c3=NULL, STRING c4=c4_4",
+ rowStrings.get(4));
+
+ KuduScanner scanner = syncClient.newScannerBuilder(table).build();
+
+ assertTrue("Scanner should have returned row", scanner.hasMoreRows());
+
+ RowResultIterator rows = scanner.nextRows();
+ final RowResult next = rows.next();
+
+ // Do negative testing on string type.
+ try {
+ next.getInt("c2");
+ fail("IllegalArgumentException was not thrown when accessing " +
+ "a string column with getInt");
+ } catch (IllegalArgumentException ignored) {}
+ }
+
+ /**
+ * Test to verify that we can write in and read back UTF8.
+ */
+ @Test(timeout = 100000)
+ public void testUTF8() throws Exception {
+ Schema schema = createManyStringsSchema();
+ syncClient.createTable(tableName, schema, createTableOptions());
+
+ KuduSession session = syncClient.newSession();
+ KuduTable table = syncClient.openTable(tableName);
+ Insert insert = table.newInsert();
+ PartialRow row = insert.getRow();
+ row.addString("key", "\u0e01\u0e02\u0e03\u0e04\u0e05\u0e06\u0e07"); // some thai
+ row.addString("c1", "\u2701\u2702\u2703\u2704\u2706"); // some icons
+
+ row.addString("c2", "hello"); // some normal chars
+ row.addString("c4", "\U0001f431"); // supplemental plane
+ session.apply(insert);
+ session.flush();
+
+ List<String> rowStrings = scanTableToStrings(table);
+ assertEquals(1, rowStrings.size());
+ assertEquals(
+ "STRING key=\u0e01\u0e02\u0e03\u0e04\u0e05\u0e06\u0e07, STRING c1=\u2701\u2702\u2703\u2704\u2706, STRING c2=hello, STRING c3=NULL, STRING c4=\U0001f431",
+ rowStrings.get(0));
+ }
+
+ /**
+ * Test inserting and retrieving binary columns.
+ */
+ @Test(timeout = 100000)
+ public void testBinaryColumns() throws Exception {
+ Schema schema = createSchemaWithBinaryColumns();
+ syncClient.createTable(tableName, schema, createTableOptions());
+
+ byte[] testArray = new byte[] {1, 2, 3, 4, 5, 6 ,7, 8, 9};
+
+ KuduSession session = syncClient.newSession();
+ KuduTable table = syncClient.openTable(tableName);
+ for (int i = 0; i < 100; i++) {
+ Insert insert = table.newInsert();
+ PartialRow row = insert.getRow();
+ row.addBinary("key", String.format("key_%02d", i).getBytes());
+ row.addString("c1", "\u2701\u2702\u2703\u2704\u2706");
+ row.addDouble("c2", i);
+ if (i % 2 == 1) {
+ row.addBinary("c3", testArray);
+ }
+ session.apply(insert);
+ if (i % 50 == 0) {
+ session.flush();
+ }
+ }
+ session.flush();
+
+ List<String> rowStrings = scanTableToStrings(table);
+ assertEquals(100, rowStrings.size());
+ for (int i = 0; i < rowStrings.size(); i++) {
+ StringBuilder expectedRow = new StringBuilder();
+ expectedRow.append(String.format("BINARY key=\"key_%02d\", STRING c1=\u2701\u2702\u2703\u2704\u2706, DOUBLE c2=%.1f,"
+ + " BINARY c3=", i, (double) i));
+ if (i % 2 == 1) {
+ expectedRow.append(Bytes.pretty(testArray));
+ } else {
+ expectedRow.append("NULL");
+ }
+ assertEquals(expectedRow.toString(), rowStrings.get(i));
+ }
+ }
+
+ /**
+ * Test inserting and retrieving timestamp columns.
+ */
+ @Test(timeout = 100000)
+ public void testTimestampColumns() throws Exception {
+ Schema schema = createSchemaWithTimestampColumns();
+ syncClient.createTable(tableName, schema, createTableOptions());
+
+ List<Long> timestamps = new ArrayList<>();
+
+ KuduSession session = syncClient.newSession();
+ KuduTable table = syncClient.openTable(tableName);
+ long lastTimestamp = 0;
+ for (int i = 0; i < 100; i++) {
+ Insert insert = table.newInsert();
+ PartialRow row = insert.getRow();
+ long timestamp = System.currentTimeMillis() * 1000;
+ while(timestamp == lastTimestamp) {
+ timestamp = System.currentTimeMillis() * 1000;
+ }
+ timestamps.add(timestamp);
+ row.addLong("key", timestamp);
+ if (i % 2 == 1) {
+ row.addLong("c1", timestamp);
+ }
+ session.apply(insert);
+ if (i % 50 == 0) {
+ session.flush();
+ }
+ lastTimestamp = timestamp;
+ }
+ session.flush();
+
+ List<String> rowStrings = scanTableToStrings(table);
+ assertEquals(100, rowStrings.size());
+ for (int i = 0; i < rowStrings.size(); i++) {
+ StringBuilder expectedRow = new StringBuilder();
+ expectedRow.append(String.format("TIMESTAMP key=%s, TIMESTAMP c1=",
+ timestampToString(timestamps.get(i))));
+ if (i % 2 == 1) {
+ expectedRow.append(timestampToString(timestamps.get(i)));
+ } else {
+ expectedRow.append("NULL");
+ }
+ assertEquals(expectedRow.toString(), rowStrings.get(i));
+ }
+ }
+
+ /**
+ * Test scanning with predicates.
+ */
+ @Test
+ public void testScanWithPredicates() throws Exception {
+ Schema schema = createManyStringsSchema();
+ syncClient.createTable(tableName, schema, createTableOptions());
+
+ 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();
+
+ assertEquals(100, scanTableToStrings(table).size());
+ assertEquals(50, scanTableToStrings(table,
+ KuduPredicate.newComparisonPredicate(schema.getColumn("key"), GREATER_EQUAL, "key_50")
+ ).size());
+ assertEquals(25, scanTableToStrings(table,
+ KuduPredicate.newComparisonPredicate(schema.getColumn("key"), GREATER, "key_74")
+ ).size());
+ assertEquals(25, scanTableToStrings(table,
+ KuduPredicate.newComparisonPredicate(schema.getColumn("key"), GREATER, "key_24"),
+ KuduPredicate.newComparisonPredicate(schema.getColumn("c1"), LESS_EQUAL, "c1_49")
+ ).size());
+ assertEquals(50, scanTableToStrings(table,
+ KuduPredicate.newComparisonPredicate(schema.getColumn("key"), GREATER, "key_24"),
+ KuduPredicate.newComparisonPredicate(schema.getColumn("key"), GREATER_EQUAL, "key_50")
+ ).size());
+ assertEquals(0, scanTableToStrings(table,
+ KuduPredicate.newComparisonPredicate(schema.getColumn("c1"), GREATER, "c1_30"),
+ KuduPredicate.newComparisonPredicate(schema.getColumn("c2"), LESS, "c2_20")
+ ).size());
+ assertEquals(0, scanTableToStrings(table,
+ // Short circuit scan
+ KuduPredicate.newComparisonPredicate(schema.getColumn("c2"), GREATER, "c2_30"),
+ KuduPredicate.newComparisonPredicate(schema.getColumn("c2"), LESS, "c2_20")
+ ).size());
+ }
+
+ /**
+ * Tests scan tokens by creating a set of scan tokens, serializing them, and
+ * then executing them in parallel with separate client instances. This
+ * simulates the normal usecase of scan tokens being created at a central
+ * planner and distributed to remote task executors.
+ */
+ @Test
+ public void testScanTokens() throws Exception {
+ Schema schema = createManyStringsSchema();
+ CreateTableOptions createOptions = new CreateTableOptions();
+ createOptions.addHashPartitions(ImmutableList.of("key"), 8);
+
+ PartialRow splitRow = schema.newPartialRow();
+ splitRow.addString("key", "key_50");
+ createOptions.addSplitRow(splitRow);
+
+ syncClient.createTable(tableName, schema, createOptions);
+
+ 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();
+
+ KuduScanToken.KuduScanTokenBuilder tokenBuilder = syncClient.newScanTokenBuilder(table);
+ tokenBuilder.setProjectedColumnIndexes(ImmutableList.<Integer>of());
+ List<KuduScanToken> tokens = tokenBuilder.build();
+ assertEquals(16, tokens.size());
+
+ final AtomicInteger count = new AtomicInteger(0);
+ List<Thread> threads = new ArrayList<>();
+ for (final KuduScanToken token : tokens) {
+ final byte[] serializedToken = token.serialize();
+ Thread thread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try (KuduClient contextClient = new KuduClient.KuduClientBuilder(masterAddresses)
+ .defaultAdminOperationTimeoutMs(DEFAULT_SLEEP)
+ .build()) {
+ KuduScanner scanner = KuduScanToken.deserializeIntoScanner(serializedToken, contextClient);
+ try {
+ int localCount = 0;
+ while (scanner.hasMoreRows()) {
+ localCount += Iterators.size(scanner.nextRows());
+ }
+ assertTrue(localCount > 0);
+ count.addAndGet(localCount);
+ } finally {
+ scanner.close();
+ }
+ } catch (Exception e) {
+ LOG.error("exception in parallel token scanner", e);
+ }
+ }
+ });
+ thread.run();
+ threads.add(thread);
+ }
+
+ for (Thread thread : threads) {
+ thread.join();
+ }
+ assertEquals(100, count.get());
+ }
+
+ /**
+ * Counts the rows in a table between two optional bounds.
+ * @param table the table to scan, must have the basic schema
+ * @param lowerBound an optional lower bound key
+ * @param upperBound an optional upper bound key
+ * @return the row count
+ * @throws Exception on error
+ */
+ private int countRowsForTestScanNonCoveredTable(KuduTable table,
+ Integer lowerBound,
+ Integer upperBound) throws Exception {
+
+ KuduScanner.KuduScannerBuilder scanBuilder = syncClient.newScannerBuilder(table);
+ if (lowerBound != null) {
+ PartialRow bound = basicSchema.newPartialRow();
+ bound.addInt(0, lowerBound);
+ scanBuilder.lowerBound(bound);
+ }
+ if (upperBound != null) {
+ PartialRow bound = basicSchema.newPartialRow();
+ bound.addInt(0, upperBound);
+ scanBuilder.exclusiveUpperBound(bound);
+ }
+
+ KuduScanner scanner = scanBuilder.build();
+ int count = 0;
+ while (scanner.hasMoreRows()) {
+ count += scanner.nextRows().getNumRows();
+ }
+ return count;
+ }
+
+ /**
+ * Tests scanning a table with non-covering range partitions.
+ */
+ @Test(timeout = 100000)
+ public void testScanNonCoveredTable() throws Exception {
+
+ Schema schema = basicSchema;
+ syncClient.createTable(tableName, schema, getBasicTableOptionsWithNonCoveredRange());
+
+ KuduSession session = syncClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);
+ KuduTable table = syncClient.openTable(tableName);
+
+ for (int key = 0; key < 100; key++) {
+ session.apply(createBasicSchemaInsert(table, key));
+ }
+ for (int key = 200; key < 300; key++) {
+ session.apply(createBasicSchemaInsert(table, key));
+ }
+ session.flush();
+ assertEquals(0, session.countPendingErrors());
+
+ assertEquals(200, countRowsForTestScanNonCoveredTable(table, null, null));
+ assertEquals(100, countRowsForTestScanNonCoveredTable(table, null, 200));
+ assertEquals(0, countRowsForTestScanNonCoveredTable(table, null, -1));
+ assertEquals(0, countRowsForTestScanNonCoveredTable(table, 120, 180));
+ assertEquals(0, countRowsForTestScanNonCoveredTable(table, 300, null));
+ }
+
+ /**
+ * Creates a local client that we auto-close while buffering one row, then makes sure that after
+ * closing that we can read the row.
+ */
+ @Test(timeout = 100000)
+ public void testAutoClose() throws Exception {
+ try (KuduClient localClient = new KuduClient.KuduClientBuilder(masterAddresses).build()) {
+ localClient.createTable(tableName, basicSchema, getBasicCreateTableOptions());
+ KuduTable table = localClient.openTable(tableName);
+ KuduSession session = localClient.newSession();
+
+ session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
+ Insert insert = createBasicSchemaInsert(table, 0);
+ session.apply(insert);
+ }
+
+ KuduTable table = syncClient.openTable(tableName);
+ AsyncKuduScanner scanner = new AsyncKuduScanner.AsyncKuduScannerBuilder(client, table).build();
+ assertEquals(1, countRowsInScan(scanner));
+ }
+
+ @Test(timeout = 100000)
+ public void testCustomNioExecutor() throws Exception {
+ long startTime = System.nanoTime();
+ final KuduClient localClient = new KuduClient.KuduClientBuilder(masterAddresses)
+ .nioExecutors(Executors.newFixedThreadPool(1), Executors.newFixedThreadPool(2))
+ .bossCount(1)
+ .workerCount(2)
+ .build();
+ long buildTime = (System.nanoTime() - startTime) / 1000000000L;
+ assertTrue("Building KuduClient is slow, maybe netty get stuck", buildTime < 3);
+ localClient.createTable(tableName, basicSchema, getBasicCreateTableOptions());
+ Thread[] threads = new Thread[4];
+ for (int t = 0; t < 4; t++) {
+ final int id = t;
+ threads[t] = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ KuduTable table = localClient.openTable(tableName);
+ KuduSession session = localClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);
+ for (int i = 0; i < 100; i++) {
+ Insert insert = createBasicSchemaInsert(table, id * 100 + i);
+ session.apply(insert);
+ }
+ session.close();
+ } catch (Exception e) {
+ fail("insert thread should not throw exception: " + e);
+ }
+ }
+ });
+ threads[t].start();
+ }
+ for (int t = 0; t< 4;t++) {
+ threads[t].join();
+ }
+ localClient.shutdown();
+ }
+
+ @Test(expected=IllegalArgumentException.class)
+ public void testNoDefaultPartitioning() throws Exception {
+ syncClient.createTable(tableName, basicSchema, new CreateTableOptions());
+ }
+}