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:29 UTC
[24/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/main/java/org/apache/kudu/client/ListTabletsRequest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsRequest.java
new file mode 100644
index 0000000..8a29b7b
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsRequest.java
@@ -0,0 +1,70 @@
+// 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.protobuf.Message;
+import org.kududb.annotations.InterfaceAudience;
+import org.kududb.tablet.Tablet;
+import org.kududb.tserver.Tserver;
+import org.kududb.tserver.TserverService;
+import org.kududb.util.Pair;
+import org.jboss.netty.buffer.ChannelBuffer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@InterfaceAudience.Private
+class ListTabletsRequest extends KuduRpc<ListTabletsResponse> {
+
+ ListTabletsRequest() {
+ super(null);
+ }
+
+ @Override
+ ChannelBuffer serialize(Message header) {
+ assert header.isInitialized();
+ final Tserver.ListTabletsRequestPB.Builder builder =
+ Tserver.ListTabletsRequestPB.newBuilder();
+ return toChannelBuffer(header, builder.build());
+ }
+
+ @Override
+ String serviceName() { return TABLET_SERVER_SERVICE_NAME; }
+
+ @Override
+ String method() {
+ return "ListTablets";
+ }
+
+ @Override
+ Pair<ListTabletsResponse, Object> deserialize(CallResponse callResponse,
+ String tsUUID) throws Exception {
+ final Tserver.ListTabletsResponsePB.Builder respBuilder =
+ Tserver.ListTabletsResponsePB.newBuilder();
+ readProtobuf(callResponse.getPBMessage(), respBuilder);
+ int serversCount = respBuilder.getStatusAndSchemaCount();
+ List<String> tablets = new ArrayList<String>(serversCount);
+ for (Tserver.ListTabletsResponsePB.StatusAndSchemaPB info
+ : respBuilder.getStatusAndSchemaList()) {
+ tablets.add(info.getTabletStatus().getTabletId());
+ }
+ ListTabletsResponse response = new ListTabletsResponse(deadlineTracker.getElapsedMillis(),
+ tsUUID, tablets);
+ return new Pair<ListTabletsResponse, Object>(
+ response, respBuilder.hasError() ? respBuilder.getError() : null);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsResponse.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsResponse.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsResponse.java
new file mode 100644
index 0000000..be2ed65
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsResponse.java
@@ -0,0 +1,40 @@
+// 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.kududb.annotations.InterfaceAudience;
+
+import java.util.List;
+
+@InterfaceAudience.Private
+public class ListTabletsResponse extends KuduRpcResponse {
+
+ private final List<String> tabletsList;
+
+ ListTabletsResponse(long ellapsedMillis, String tsUUID, List<String> tabletsList) {
+ super(ellapsedMillis, tsUUID);
+ this.tabletsList = tabletsList;
+ }
+
+ /**
+ * Get the list of tablets as specified in the request.
+ * @return a list of tablet uuids
+ */
+ public List<String> getTabletsList() {
+ return tabletsList;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/main/java/org/apache/kudu/client/LocatedTablet.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/LocatedTablet.java b/java/kudu-client/src/main/java/org/apache/kudu/client/LocatedTablet.java
new file mode 100644
index 0000000..67934db
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/LocatedTablet.java
@@ -0,0 +1,132 @@
+/*
+ *
+ * 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 java.util.List;
+
+import org.kududb.annotations.InterfaceAudience;
+import org.kududb.annotations.InterfaceStability;
+import org.kududb.consensus.Metadata.RaftPeerPB.Role;
+import org.kududb.master.Master.TabletLocationsPB.ReplicaPB;
+
+/**
+ * Information about the locations of tablets in a Kudu table.
+ * This should be treated as immutable data (it does not reflect
+ * any updates the client may have heard since being constructed).
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class LocatedTablet {
+ private final Partition partition;
+ private final byte[] tabletId;
+
+ private final List<Replica> replicas;
+
+ LocatedTablet(AsyncKuduClient.RemoteTablet tablet) {
+ partition = tablet.getPartition();
+ tabletId = tablet.getTabletIdAsBytes();
+ replicas = tablet.getReplicas();
+ }
+
+ public List<Replica> getReplicas() {
+ return replicas;
+ }
+
+ public Partition getPartition() {
+ return partition;
+ }
+
+ /**
+ * DEPRECATED: use {@link #getPartition()}
+ */
+ @Deprecated
+ public byte[] getStartKey() {
+ return getPartition().getPartitionKeyStart();
+ }
+
+ /**
+ * DEPRECATED: use {@link #getPartition()}
+ */
+ @Deprecated()
+ public byte[] getEndKey() {
+ return getPartition().getPartitionKeyEnd();
+ }
+
+ public byte[] getTabletId() {
+ return tabletId;
+ }
+
+ /**
+ * Return the current leader, or null if there is none.
+ */
+ public Replica getLeaderReplica() {
+ return getOneOfRoleOrNull(Role.LEADER);
+ }
+
+ /**
+ * Return the first occurrence for the given role, or null if there is none.
+ */
+ private Replica getOneOfRoleOrNull(Role role) {
+ for (Replica r : replicas) {
+ if (r.getRole() == role.toString()) return r;
+ }
+ return null;
+ }
+
+ @Override
+ public String toString() {
+ return Bytes.pretty(tabletId) + " " + partition.toString();
+ }
+
+ /**
+ * One of the replicas of the tablet.
+ */
+ @InterfaceAudience.Public
+ @InterfaceStability.Evolving
+ public static class Replica {
+ private final ReplicaPB pb;
+
+ Replica(ReplicaPB pb) {
+ this.pb = pb;
+ }
+
+ public String getRpcHost() {
+ if (pb.getTsInfo().getRpcAddressesList().isEmpty()) {
+ return null;
+ }
+ return pb.getTsInfo().getRpcAddressesList().get(0).getHost();
+ }
+
+ public Integer getRpcPort() {
+ if (pb.getTsInfo().getRpcAddressesList().isEmpty()) {
+ return null;
+ }
+ return pb.getTsInfo().getRpcAddressesList().get(0).getPort();
+ }
+
+ public String getRole() {
+ return pb.getRole().toString();
+ }
+
+ public String toString() {
+ return pb.toString();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/main/java/org/apache/kudu/client/NoLeaderMasterFoundException.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/NoLeaderMasterFoundException.java b/java/kudu-client/src/main/java/org/apache/kudu/client/NoLeaderMasterFoundException.java
new file mode 100644
index 0000000..1cde694
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/NoLeaderMasterFoundException.java
@@ -0,0 +1,37 @@
+// 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.kududb.annotations.InterfaceAudience;
+import org.kududb.annotations.InterfaceStability;
+
+import java.util.List;
+
+/**
+ * Indicates that the request failed because we couldn't find a leader master server.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+final class NoLeaderMasterFoundException extends RecoverableException {
+
+ NoLeaderMasterFoundException(Status status) {
+ super(status);
+ }
+ NoLeaderMasterFoundException(Status status, Exception cause) {
+ super(status, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/main/java/org/apache/kudu/client/NonCoveredRangeCache.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/NonCoveredRangeCache.java b/java/kudu-client/src/main/java/org/apache/kudu/client/NonCoveredRangeCache.java
new file mode 100644
index 0000000..1c3b024
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/NonCoveredRangeCache.java
@@ -0,0 +1,104 @@
+// 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.Joiner;
+import com.google.common.primitives.UnsignedBytes;
+
+import java.util.Comparator;
+import java.util.Map;
+import java.util.concurrent.ConcurrentNavigableMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import javax.annotation.concurrent.ThreadSafe;
+
+import org.kududb.annotations.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A cache of the non-covered range partitions in a Kudu table.
+ *
+ * Currently entries are never invalidated from the cache.
+ */
+@ThreadSafe
+@InterfaceAudience.Private
+class NonCoveredRangeCache {
+ private static final Logger LOG = LoggerFactory.getLogger(NonCoveredRangeCache.class);
+ private static final Comparator<byte[]> COMPARATOR = UnsignedBytes.lexicographicalComparator();
+
+ private final ConcurrentNavigableMap<byte[], byte[]> nonCoveredRanges =
+ new ConcurrentSkipListMap<>(COMPARATOR);
+
+ /**
+ * Retrieves a non-covered range from the cache.
+ *
+ * The pair contains the inclusive start partition key and the exclusive end
+ * partition key containing the provided partition key. If there is no such
+ * cached range, null is returned.
+ *
+ * @param partitionKey the partition key to lookup in the cache
+ * @return the non covered range, or null
+ */
+ public Map.Entry<byte[], byte[]> getNonCoveredRange(byte[] partitionKey) {
+ Map.Entry<byte[], byte[]> range = nonCoveredRanges.floorEntry(partitionKey);
+ if (range == null ||
+ (range.getValue().length != 0 && COMPARATOR.compare(partitionKey, range.getValue()) >= 0)) {
+ return null;
+ } else {
+ return range;
+ }
+ }
+
+ /**
+ * Adds a non-covered range to the cache.
+ *
+ * @param startPartitionKey the inclusive start partition key of the non-covered range
+ * @param endPartitionKey the exclusive end partition key of the non-covered range
+ */
+ public void addNonCoveredRange(byte[] startPartitionKey, byte[] endPartitionKey) {
+ if (startPartitionKey == null || endPartitionKey == null) {
+ throw new IllegalArgumentException("Non-covered partition range keys may not be null");
+ }
+ // Concurrent additions of the same non-covered range key are handled by
+ // serializing puts through the concurrent map.
+ if (nonCoveredRanges.put(startPartitionKey, endPartitionKey) == null) {
+ LOG.info("Discovered non-covered partition range [{}, {})",
+ Bytes.hex(startPartitionKey), Bytes.hex(endPartitionKey));
+ }
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append('[');
+ boolean isFirst = true;
+ for (Map.Entry<byte[], byte[]> range : nonCoveredRanges.entrySet()) {
+ if (isFirst) {
+ isFirst = false;
+ } else {
+ sb.append(", ");
+ }
+ sb.append('[');
+ sb.append(range.getKey().length == 0 ? "<start>" : Bytes.hex(range.getKey()));
+ sb.append(", ");
+ sb.append(range.getValue().length == 0 ? "<end>" : Bytes.hex(range.getValue()));
+ sb.append(')');
+ }
+ sb.append(']');
+ return sb.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/main/java/org/apache/kudu/client/NonCoveredRangeException.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/NonCoveredRangeException.java b/java/kudu-client/src/main/java/org/apache/kudu/client/NonCoveredRangeException.java
new file mode 100644
index 0000000..b704441
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/NonCoveredRangeException.java
@@ -0,0 +1,51 @@
+// 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.kududb.annotations.InterfaceAudience;
+import org.kududb.annotations.InterfaceStability;
+
+/**
+ * Exception indicating that an operation attempted to access a non-covered range partition.
+ */
+@InterfaceAudience.Private
+class NonCoveredRangeException extends NonRecoverableException {
+ private final byte[] nonCoveredRangeStart;
+ private final byte[] nonCoveredRangeEnd;
+
+ public NonCoveredRangeException(byte[] nonCoveredRangeStart, byte[] nonCoveredRangeEnd) {
+ super(Status.NotFound("non-covered range"));
+ this.nonCoveredRangeStart = nonCoveredRangeStart;
+ this.nonCoveredRangeEnd = nonCoveredRangeEnd;
+ }
+
+ byte[] getNonCoveredRangeStart() {
+ return nonCoveredRangeStart;
+ }
+
+ byte[] getNonCoveredRangeEnd() {
+ return nonCoveredRangeEnd;
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "NonCoveredRangeException([%s, %s))",
+ nonCoveredRangeStart.length == 0 ? "<start>" : Bytes.hex(nonCoveredRangeStart),
+ nonCoveredRangeEnd.length == 0 ? "<end>" : Bytes.hex(nonCoveredRangeEnd));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/main/java/org/apache/kudu/client/NonRecoverableException.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/NonRecoverableException.java b/java/kudu-client/src/main/java/org/apache/kudu/client/NonRecoverableException.java
new file mode 100644
index 0000000..7bcb81d
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/NonRecoverableException.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright (C) 2010-2012 The Async HBase Authors. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ * - Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ * - Redistributions in binary form must reproduce the above copyright notice,
+ * this list of conditions and the following disclaimer in the documentation
+ * and/or other materials provided with the distribution.
+ * - Neither the name of the StumbleUpon nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+package org.kududb.client;
+
+import org.kududb.annotations.InterfaceAudience;
+import org.kududb.annotations.InterfaceStability;
+
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+@SuppressWarnings("serial")
+class NonRecoverableException extends KuduException {
+
+ /**
+ * Constructor.
+ * @param status status object containing the reason for the exception
+ * trace.
+ */
+ NonRecoverableException(Status status) {
+ super(status);
+ }
+
+ /**
+ * Constructor.
+ * @param status status object containing the reason for the exception
+ * @param cause The exception that caused this one to be thrown.
+ */
+ NonRecoverableException(Status status, Throwable cause) {
+ super(status, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java b/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
new file mode 100644
index 0000000..e27c222
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
@@ -0,0 +1,345 @@
+// 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.google.protobuf.ByteString;
+import com.google.protobuf.Message;
+import com.google.protobuf.ZeroCopyLiteralByteString;
+
+import org.kududb.ColumnSchema;
+import org.kududb.Schema;
+import org.kududb.Type;
+import org.kududb.WireProtocol;
+import org.kududb.WireProtocol.RowOperationsPB;
+import org.kududb.annotations.InterfaceAudience;
+import org.kududb.annotations.InterfaceStability;
+import org.kududb.client.Statistics.Statistic;
+import org.kududb.client.Statistics.TabletStatistics;
+import org.kududb.tserver.Tserver;
+import org.kududb.util.Pair;
+import org.kududb.util.Slice;
+import org.jboss.netty.buffer.ChannelBuffer;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Base class for the RPCs that related to WriteRequestPB. It contains almost all the logic
+ * and knows how to serialize its child classes.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public abstract class Operation extends KuduRpc<OperationResponse> {
+ /**
+ * This size will be set when serialize is called. It stands for the size of the row in this
+ * operation.
+ */
+ private long rowOperationSizeBytes = 0;
+
+ enum ChangeType {
+ INSERT((byte)RowOperationsPB.Type.INSERT.getNumber()),
+ UPDATE((byte)RowOperationsPB.Type.UPDATE.getNumber()),
+ DELETE((byte)RowOperationsPB.Type.DELETE.getNumber()),
+ SPLIT_ROWS((byte)RowOperationsPB.Type.SPLIT_ROW.getNumber()),
+ UPSERT((byte)RowOperationsPB.Type.UPSERT.getNumber()),
+ RANGE_LOWER_BOUND((byte) RowOperationsPB.Type.RANGE_LOWER_BOUND.getNumber()),
+ RANGE_UPPER_BOUND((byte) RowOperationsPB.Type.RANGE_UPPER_BOUND.getNumber());
+
+ ChangeType(byte encodedByte) {
+ this.encodedByte = encodedByte;
+ }
+
+ byte toEncodedByte() {
+ return encodedByte;
+ }
+
+ /** The byte used to encode this in a RowOperationsPB */
+ private byte encodedByte;
+ }
+
+ static final String METHOD = "Write";
+
+ private final PartialRow row;
+
+ /** See {@link SessionConfiguration#setIgnoreAllDuplicateRows(boolean)} */
+ boolean ignoreAllDuplicateRows = false;
+
+ /**
+ * Package-private constructor. Subclasses need to be instantiated via AsyncKuduSession
+ * @param table table with the schema to use for this operation
+ */
+ Operation(KuduTable table) {
+ super(table);
+ this.row = table.getSchema().newPartialRow();
+ }
+
+ /** See {@link SessionConfiguration#setIgnoreAllDuplicateRows(boolean)} */
+ void setIgnoreAllDuplicateRows(boolean ignoreAllDuplicateRows) {
+ this.ignoreAllDuplicateRows = ignoreAllDuplicateRows;
+ }
+
+ /**
+ * Classes extending Operation need to have a specific ChangeType
+ * @return Operation's ChangeType
+ */
+ abstract ChangeType getChangeType();
+
+ /**
+ * Returns the size in bytes of this operation's row after serialization.
+ * @return size in bytes
+ * @throws IllegalStateException thrown if this RPC hasn't been serialized eg sent to a TS
+ */
+ long getRowOperationSizeBytes() {
+ if (this.rowOperationSizeBytes == 0) {
+ throw new IllegalStateException("This row hasn't been serialized yet");
+ }
+ return this.rowOperationSizeBytes;
+ }
+
+ @Override
+ String serviceName() { return TABLET_SERVER_SERVICE_NAME; }
+
+ @Override
+ String method() {
+ return METHOD;
+ }
+
+ @Override
+ ChannelBuffer serialize(Message header) {
+ final Tserver.WriteRequestPB.Builder builder =
+ createAndFillWriteRequestPB(ImmutableList.of(this));
+ this.rowOperationSizeBytes = builder.getRowOperations().getRows().size()
+ + builder.getRowOperations().getIndirectData().size();
+ builder.setTabletId(ZeroCopyLiteralByteString.wrap(getTablet().getTabletIdAsBytes()));
+ builder.setExternalConsistencyMode(this.externalConsistencyMode.pbVersion());
+ if (this.propagatedTimestamp != AsyncKuduClient.NO_TIMESTAMP) {
+ builder.setPropagatedTimestamp(this.propagatedTimestamp);
+ }
+ return toChannelBuffer(header, builder.build());
+ }
+
+ @Override
+ Pair<OperationResponse, Object> deserialize(CallResponse callResponse,
+ String tsUUID) throws Exception {
+ Tserver.WriteResponsePB.Builder builder = Tserver.WriteResponsePB.newBuilder();
+ readProtobuf(callResponse.getPBMessage(), builder);
+ Tserver.WriteResponsePB.PerRowErrorPB error = null;
+ if (builder.getPerRowErrorsCount() != 0) {
+ error = builder.getPerRowErrors(0);
+ if (ignoreAllDuplicateRows &&
+ error.getError().getCode() == WireProtocol.AppStatusPB.ErrorCode.ALREADY_PRESENT) {
+ error = null;
+ }
+ }
+ OperationResponse response = new OperationResponse(deadlineTracker.getElapsedMillis(), tsUUID,
+ builder.getTimestamp(), this, error);
+ return new Pair<OperationResponse, Object>(
+ response, builder.hasError() ? builder.getError() : null);
+ }
+
+ @Override
+ public byte[] partitionKey() {
+ return this.getTable().getPartitionSchema().encodePartitionKey(row);
+ }
+
+ @Override
+ boolean isRequestTracked() {
+ return true;
+ }
+
+ /**
+ * Get the underlying row to modify.
+ * @return a partial row that will be sent with this Operation
+ */
+ public PartialRow getRow() {
+ return this.row;
+ }
+
+ @Override
+ void updateStatistics(Statistics statistics, OperationResponse response) {
+ Slice tabletId = this.getTablet().getTabletId();
+ String tableName = this.getTable().getName();
+ TabletStatistics tabletStatistics = statistics.getTabletStatistics(tableName, tabletId);
+ if (response == null) {
+ tabletStatistics.incrementStatistic(Statistic.OPS_ERRORS, 1);
+ tabletStatistics.incrementStatistic(Statistic.RPC_ERRORS, 1);
+ return;
+ }
+ tabletStatistics.incrementStatistic(Statistic.WRITE_RPCS, 1);
+ if (response.hasRowError()) {
+ // If ignoreAllDuplicateRows is set, the already_present exception will be
+ // discarded and wont't be recorded here
+ tabletStatistics.incrementStatistic(Statistic.OPS_ERRORS, 1);
+ } else {
+ tabletStatistics.incrementStatistic(Statistic.WRITE_OPS, 1);
+ }
+ tabletStatistics.incrementStatistic(Statistic.BYTES_WRITTEN, getRowOperationSizeBytes());
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder(super.toString());
+ sb.append(" row_key=");
+ sb.append(row.stringifyRowKey());
+ return sb.toString();
+ }
+
+ /**
+ * Helper method that puts a list of Operations together into a WriteRequestPB.
+ * @param operations The list of ops to put together in a WriteRequestPB
+ * @return A fully constructed WriteRequestPB containing the passed rows, or
+ * null if no rows were passed.
+ */
+ static Tserver.WriteRequestPB.Builder createAndFillWriteRequestPB(List<Operation> operations) {
+ if (operations == null || operations.isEmpty()) return null;
+ Schema schema = operations.get(0).table.getSchema();
+ RowOperationsPB rowOps = new OperationsEncoder().encodeOperations(operations);
+ if (rowOps == null) return null;
+
+ Tserver.WriteRequestPB.Builder requestBuilder = Tserver.WriteRequestPB.newBuilder();
+ requestBuilder.setSchema(ProtobufHelper.schemaToPb(schema));
+ requestBuilder.setRowOperations(rowOps);
+ return requestBuilder;
+ }
+
+ static class OperationsEncoder {
+ private Schema schema;
+ private ByteBuffer rows;
+ // We're filling this list as we go through the operations in encodeRow() and at the same time
+ // compute the total size, which will be used to right-size the array in toPB().
+ private List<ByteBuffer> indirect;
+ private long indirectWrittenBytes;
+
+ /**
+ * Initializes the state of the encoder based on the schema and number of operations to encode.
+ *
+ * @param schema the schema of the table which the operations belong to.
+ * @param numOperations the number of operations.
+ */
+ private void init(Schema schema, int numOperations) {
+ this.schema = schema;
+
+ // Set up the encoded data.
+ // Estimate a maximum size for the data. This is conservative, but avoids
+ // having to loop through all the operations twice.
+ final int columnBitSetSize = Bytes.getBitSetSize(schema.getColumnCount());
+ int sizePerRow = 1 /* for the op type */ + schema.getRowSize() + columnBitSetSize;
+ if (schema.hasNullableColumns()) {
+ // nullsBitSet is the same size as the columnBitSet
+ sizePerRow += columnBitSetSize;
+ }
+
+ // TODO: would be more efficient to use a buffer which "chains" smaller allocations
+ // instead of a doubling buffer like BAOS.
+ this.rows = ByteBuffer.allocate(sizePerRow * numOperations)
+ .order(ByteOrder.LITTLE_ENDIAN);
+ this.indirect = new ArrayList<>(schema.getVarLengthColumnCount() * numOperations);
+ }
+
+ /**
+ * Builds the row operations protobuf message with encoded operations.
+ * @return the row operations protobuf message.
+ */
+ private RowOperationsPB toPB() {
+ RowOperationsPB.Builder rowOpsBuilder = RowOperationsPB.newBuilder();
+
+ // TODO: we could implement a ZeroCopy approach here by subclassing LiteralByteString.
+ // We have ZeroCopyLiteralByteString, but that only supports an entire array. Here
+ // we've only partially filled in rows.array(), so we have to make the extra copy.
+ rows.limit(rows.position());
+ rows.flip();
+ rowOpsBuilder.setRows(ByteString.copyFrom(rows));
+ if (indirect.size() > 0) {
+ // TODO: same as above, we could avoid a copy here by using an implementation that allows
+ // zero-copy on a slice of an array.
+ byte[] indirectData = new byte[(int)indirectWrittenBytes];
+ int offset = 0;
+ for (ByteBuffer bb : indirect) {
+ int bbSize = bb.remaining();
+ bb.get(indirectData, offset, bbSize);
+ offset += bbSize;
+ }
+ rowOpsBuilder.setIndirectData(ZeroCopyLiteralByteString.wrap(indirectData));
+ }
+ return rowOpsBuilder.build();
+ }
+
+ private void encodeRow(PartialRow row, ChangeType type) {
+ rows.put(type.toEncodedByte());
+ rows.put(Bytes.fromBitSet(row.getColumnsBitSet(), schema.getColumnCount()));
+ if (schema.hasNullableColumns()) {
+ rows.put(Bytes.fromBitSet(row.getNullsBitSet(), schema.getColumnCount()));
+ }
+ int colIdx = 0;
+ byte[] rowData = row.getRowAlloc();
+ int currentRowOffset = 0;
+ for (ColumnSchema col : row.getSchema().getColumns()) {
+ // Keys should always be specified, maybe check?
+ if (row.isSet(colIdx) && !row.isSetToNull(colIdx)) {
+ if (col.getType() == Type.STRING || col.getType() == Type.BINARY) {
+ ByteBuffer varLengthData = row.getVarLengthData().get(colIdx);
+ varLengthData.reset();
+ rows.putLong(indirectWrittenBytes);
+ int bbSize = varLengthData.remaining();
+ rows.putLong(bbSize);
+ indirect.add(varLengthData);
+ indirectWrittenBytes += bbSize;
+ } else {
+ // This is for cols other than strings
+ rows.put(rowData, currentRowOffset, col.getType().getSize());
+ }
+ }
+ currentRowOffset += col.getType().getSize();
+ colIdx++;
+ }
+ }
+
+ public RowOperationsPB encodeOperations(List<Operation> operations) {
+ if (operations == null || operations.isEmpty()) return null;
+ init(operations.get(0).table.getSchema(), operations.size());
+ for (Operation operation : operations) {
+ encodeRow(operation.row, operation.getChangeType());
+ }
+ return toPB();
+ }
+
+ public RowOperationsPB encodeSplitRowsRangeBounds(List<PartialRow> splitRows,
+ List<Pair<PartialRow, PartialRow>> rangeBounds) {
+ if (splitRows.isEmpty() && rangeBounds.isEmpty()) {
+ return null;
+ }
+
+ Schema schema = splitRows.isEmpty() ? rangeBounds.get(0).getFirst().getSchema()
+ : splitRows.get(0).getSchema();
+ init(schema, splitRows.size() + 2 * rangeBounds.size());
+
+ for (PartialRow row : splitRows) {
+ encodeRow(row, ChangeType.SPLIT_ROWS);
+ }
+
+ for (Pair<PartialRow, PartialRow> bound : rangeBounds) {
+ encodeRow(bound.getFirst(), ChangeType.RANGE_LOWER_BOUND);
+ encodeRow(bound.getSecond(), ChangeType.RANGE_UPPER_BOUND);
+ }
+
+ return toPB();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/main/java/org/apache/kudu/client/OperationResponse.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/OperationResponse.java b/java/kudu-client/src/main/java/org/apache/kudu/client/OperationResponse.java
new file mode 100644
index 0000000..bf707ce
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/OperationResponse.java
@@ -0,0 +1,111 @@
+// 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.kududb.annotations.InterfaceAudience;
+import org.kududb.annotations.InterfaceStability;
+import org.kududb.tserver.Tserver;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class OperationResponse extends KuduRpcResponse {
+
+ private final long writeTimestamp;
+ private final RowError rowError;
+ private final Operation operation;
+
+ /**
+ * Package-private constructor to build an OperationResponse with a row error in the pb format.
+ * @param elapsedMillis time in milliseconds since RPC creation to now
+ * @param writeTimestamp HT's write timestamp
+ * @param operation the operation that created this response
+ * @param errorPB a row error in pb format, can be null
+ */
+ OperationResponse(long elapsedMillis, String tsUUID, long writeTimestamp,
+ Operation operation, Tserver.WriteResponsePB.PerRowErrorPB errorPB) {
+ super(elapsedMillis, tsUUID);
+ this.writeTimestamp = writeTimestamp;
+ this.rowError = errorPB == null ? null : RowError.fromRowErrorPb(errorPB, operation, tsUUID);
+ this.operation = operation;
+ }
+
+ /**
+ * Package-private constructor to build an OperationResponse with a row error.
+ * @param elapsedMillis time in milliseconds since RPC creation to now
+ * @param writeTimestamp HT's write timestamp
+ * @param operation the operation that created this response
+ * @param rowError a parsed row error, can be null
+ */
+ OperationResponse(long elapsedMillis, String tsUUID, long writeTimestamp,
+ Operation operation, RowError rowError) {
+ super(elapsedMillis, tsUUID);
+ this.writeTimestamp = writeTimestamp;
+ this.rowError = rowError;
+ this.operation = operation;
+ }
+
+ /**
+ * Utility method that collects all the row errors from the given list of responses.
+ * @param responses a list of operation responses to collect the row errors from
+ * @return a combined list of row errors
+ */
+ public static List<RowError> collectErrors(List<OperationResponse> responses) {
+ List<RowError> errors = new ArrayList<>(responses.size());
+ for (OperationResponse resp : responses) {
+ if (resp.hasRowError()) {
+ errors.add(resp.getRowError());
+ }
+ }
+ return errors;
+ }
+
+ /**
+ * Gives the write timestamp that was returned by the Tablet Server.
+ * @return a timestamp in milliseconds, 0 if the external consistency mode set
+ * in AsyncKuduSession wasn't CLIENT_PROPAGATED, or if the operation failed.
+ */
+ public long getWriteTimestamp() {
+ return writeTimestamp;
+ }
+
+ /**
+ * Returns a row error. If {@link #hasRowError()} returns false, then this method returns null.
+ * @return a row error, or null if the operation was successful
+ */
+ public RowError getRowError() {
+ return rowError;
+ }
+
+ /**
+ * Tells if this operation response contains a row error.
+ * @return true if this operation response has errors, else false
+ */
+ public boolean hasRowError() {
+ return rowError != null;
+ }
+
+ /**
+ * Returns the operation associated with this response.
+ * @return an operation, cannot be null
+ */
+ Operation getOperation() {
+ return operation;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/main/java/org/apache/kudu/client/PartialRow.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/PartialRow.java b/java/kudu-client/src/main/java/org/apache/kudu/client/PartialRow.java
new file mode 100644
index 0000000..b5f3069
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/PartialRow.java
@@ -0,0 +1,626 @@
+// 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 java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.List;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import org.kududb.ColumnSchema;
+import org.kududb.Schema;
+import org.kududb.Type;
+import org.kududb.annotations.InterfaceAudience;
+import org.kududb.annotations.InterfaceStability;
+
+/**
+ * Class used to represent parts of a row along with its schema.<p>
+ *
+ * Values can be replaced as often as needed, but once the enclosing {@link Operation} is applied
+ * then they cannot be changed again. This means that a PartialRow cannot be reused.<p>
+ *
+ * Each PartialRow is backed by an byte array where all the cells (except strings and binary data)
+ * are written. The others are kept in a List.<p>
+ *
+ * This class isn't thread-safe.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class PartialRow {
+
+ private final Schema schema;
+
+ // Variable length data. If string, will be UTF-8 encoded. Elements of this list _must_ have a
+ // mark that we can reset() to. Readers of these fields (encoders, etc) must call reset() before
+ // attempting to read these values.
+ private final List<ByteBuffer> varLengthData;
+ private final byte[] rowAlloc;
+
+ private final BitSet columnsBitSet;
+ private final BitSet nullsBitSet;
+
+ private boolean frozen = false;
+
+ /**
+ * This is not a stable API, prefer using {@link Schema#newPartialRow()}
+ * to create a new partial row.
+ * @param schema the schema to use for this row
+ */
+ public PartialRow(Schema schema) {
+ this.schema = schema;
+ this.columnsBitSet = new BitSet(this.schema.getColumnCount());
+ this.nullsBitSet = schema.hasNullableColumns() ?
+ new BitSet(this.schema.getColumnCount()) : null;
+ this.rowAlloc = new byte[schema.getRowSize()];
+ // Pre-fill the array with nulls. We'll only replace cells that have varlen values.
+ this.varLengthData = Arrays.asList(new ByteBuffer[this.schema.getColumnCount()]);
+ }
+
+ /**
+ * Creates a new partial row by deep-copying the data-fields of the provided partial row.
+ * @param row the partial row to copy
+ */
+ PartialRow(PartialRow row) {
+ this.schema = row.schema;
+
+ this.varLengthData = Lists.newArrayListWithCapacity(row.varLengthData.size());
+ for (ByteBuffer data: row.varLengthData) {
+ if (data == null) {
+ this.varLengthData.add(null);
+ } else {
+ data.reset();
+ // Deep copy the ByteBuffer.
+ ByteBuffer clone = ByteBuffer.allocate(data.remaining());
+ clone.put(data);
+ clone.flip();
+
+ clone.mark(); // We always expect a mark.
+ this.varLengthData.add(clone);
+ }
+ }
+
+ this.rowAlloc = row.rowAlloc.clone();
+ this.columnsBitSet = (BitSet) row.columnsBitSet.clone();
+ this.nullsBitSet = row.nullsBitSet == null ? null : (BitSet) row.nullsBitSet.clone();
+ }
+
+ /**
+ * Add a boolean for the specified column.
+ * @param columnIndex the column's index in the schema
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addBoolean(int columnIndex, boolean val) {
+ checkColumn(schema.getColumnByIndex(columnIndex), Type.BOOL);
+ rowAlloc[getPositionInRowAllocAndSetBitSet(columnIndex)] = (byte) (val ? 1 : 0);
+ }
+
+ /**
+ * Add a boolean for the specified column.
+ * @param columnName Name of the column
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addBoolean(String columnName, boolean val) {
+ addBoolean(schema.getColumnIndex(columnName), val);
+ }
+
+ /**
+ * Add a byte for the specified column.
+ * @param columnIndex the column's index in the schema
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addByte(int columnIndex, byte val) {
+ checkColumn(schema.getColumnByIndex(columnIndex), Type.INT8);
+ rowAlloc[getPositionInRowAllocAndSetBitSet(columnIndex)] = val;
+ }
+
+ /**
+ * Add a byte for the specified column.
+ * @param columnName Name of the column
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addByte(String columnName, byte val) {
+ addByte(schema.getColumnIndex(columnName), val);
+ }
+
+ /**
+ * Add a short for the specified column.
+ * @param columnIndex the column's index in the schema
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addShort(int columnIndex, short val) {
+ checkColumn(schema.getColumnByIndex(columnIndex), Type.INT16);
+ Bytes.setShort(rowAlloc, val, getPositionInRowAllocAndSetBitSet(columnIndex));
+ }
+
+ /**
+ * Add a short for the specified column.
+ * @param columnName Name of the column
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addShort(String columnName, short val) {
+ addShort(schema.getColumnIndex(columnName), val);
+ }
+
+ /**
+ * Add an int for the specified column.
+ * @param columnIndex the column's index in the schema
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addInt(int columnIndex, int val) {
+ checkColumn(schema.getColumnByIndex(columnIndex), Type.INT32);
+ Bytes.setInt(rowAlloc, val, getPositionInRowAllocAndSetBitSet(columnIndex));
+ }
+
+ /**
+ * Add an int for the specified column.
+ * @param columnName Name of the column
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addInt(String columnName, int val) {
+ addInt(schema.getColumnIndex(columnName), val);
+ }
+
+ /**
+ * Add an long for the specified column.
+ * @param columnIndex the column's index in the schema
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addLong(int columnIndex, long val) {
+ checkColumn(schema.getColumnByIndex(columnIndex), Type.INT64, Type.TIMESTAMP);
+ Bytes.setLong(rowAlloc, val, getPositionInRowAllocAndSetBitSet(columnIndex));
+ }
+
+ /**
+ * Add an long for the specified column.
+ *
+ * If this is a TIMESTAMP column, the long value provided should be the number of microseconds
+ * between a given time and January 1, 1970 UTC.
+ * For example, to encode the current time, use setLong(System.currentTimeMillis() * 1000);
+ *
+ * @param columnName Name of the column
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addLong(String columnName, long val) {
+ addLong(schema.getColumnIndex(columnName), val);
+ }
+
+ /**
+ * Add an float for the specified column.
+ * @param columnIndex the column's index in the schema
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addFloat(int columnIndex, float val) {
+ checkColumn(schema.getColumnByIndex(columnIndex), Type.FLOAT);
+ Bytes.setFloat(rowAlloc, val, getPositionInRowAllocAndSetBitSet(columnIndex));
+ }
+
+ /**
+ * Add an float for the specified column.
+ * @param columnName Name of the column
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addFloat(String columnName, float val) {
+ addFloat(schema.getColumnIndex(columnName), val);
+ }
+
+ /**
+ * Add an double for the specified column.
+ * @param columnIndex the column's index in the schema
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addDouble(int columnIndex, double val) {
+ checkColumn(schema.getColumnByIndex(columnIndex), Type.DOUBLE);
+ Bytes.setDouble(rowAlloc, val, getPositionInRowAllocAndSetBitSet(columnIndex));
+ }
+
+ /**
+ * Add an double for the specified column.
+ * @param columnName Name of the column
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addDouble(String columnName, double val) {
+ addDouble(schema.getColumnIndex(columnName), val);
+ }
+
+ /**
+ * Add a String for the specified column.
+ * @param columnIndex the column's index in the schema
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addString(int columnIndex, String val) {
+ addStringUtf8(columnIndex, Bytes.fromString(val));
+ }
+
+ /**
+ * Add a String for the specified column.
+ * @param columnName Name of the column
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addString(String columnName, String val) {
+ addStringUtf8(columnName, Bytes.fromString(val));
+ }
+
+ /**
+ * Add a String for the specified value, encoded as UTF8.
+ * Note that the provided value must not be mutated after this.
+ * @param columnIndex the column's index in the schema
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addStringUtf8(int columnIndex, byte[] val) {
+ // TODO: use Utf8.isWellFormed from Guava 16 to verify that
+ // the user isn't putting in any garbage data.
+ checkColumn(schema.getColumnByIndex(columnIndex), Type.STRING);
+ addVarLengthData(columnIndex, val);
+ }
+
+ /**
+ * Add a String for the specified value, encoded as UTF8.
+ * Note that the provided value must not be mutated after this.
+ * @param columnName Name of the column
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ *
+ */
+ public void addStringUtf8(String columnName, byte[] val) {
+ addStringUtf8(schema.getColumnIndex(columnName), val);
+ }
+
+ /**
+ * Add binary data with the specified value.
+ * Note that the provided value must not be mutated after this.
+ * @param columnIndex the column's index in the schema
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addBinary(int columnIndex, byte[] val) {
+ checkColumn(schema.getColumnByIndex(columnIndex), Type.BINARY);
+ addVarLengthData(columnIndex, val);
+ }
+
+ /**
+ * Add binary data with the specified value, from the current ByteBuffer's position to its limit.
+ * This method duplicates the ByteBuffer but doesn't copy the data. This means that the wrapped
+ * data must not be mutated after this.
+ * @param columnIndex the column's index in the schema
+ * @param value byte buffer to get the value from
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addBinary(int columnIndex, ByteBuffer value) {
+ checkColumn(schema.getColumnByIndex(columnIndex), Type.BINARY);
+ addVarLengthData(columnIndex, value);
+ }
+
+ /**
+ * Add binary data with the specified value.
+ * Note that the provided value must not be mutated after this.
+ * @param columnName Name of the column
+ * @param val value to add
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addBinary(String columnName, byte[] val) {
+ addBinary(schema.getColumnIndex(columnName), val);
+ }
+
+ /**
+ * Add binary data with the specified value, from the current ByteBuffer's position to its limit.
+ * This method duplicates the ByteBuffer but doesn't copy the data. This means that the wrapped
+ * data must not be mutated after this.
+ * @param columnName Name of the column
+ * @param value byte buffer to get the value from
+ * @throws IllegalArgumentException if the column doesn't exist or if the value doesn't match
+ * the column's type
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void addBinary(String columnName, ByteBuffer value) {
+ addBinary(schema.getColumnIndex(columnName), value);
+ }
+
+ private void addVarLengthData(int columnIndex, byte[] val) {
+ addVarLengthData(columnIndex, ByteBuffer.wrap(val));
+ }
+
+ private void addVarLengthData(int columnIndex, ByteBuffer val) {
+ // A duplicate will copy all the original's metadata but still point to the same content.
+ ByteBuffer duplicate = val.duplicate();
+ // Mark the current position so we can reset to it.
+ duplicate.mark();
+
+ varLengthData.set(columnIndex, duplicate);
+ // Set the usage bit but we don't care where it is.
+ getPositionInRowAllocAndSetBitSet(columnIndex);
+ // We don't set anything in row alloc, it will be managed at encoding time.
+ }
+
+ /**
+ * Set the specified column to null
+ * @param columnIndex the column's index in the schema
+ * @throws IllegalArgumentException if the column doesn't exist or cannot be set to null
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void setNull(int columnIndex) {
+ setNull(this.schema.getColumnByIndex(columnIndex));
+ }
+
+ /**
+ * Set the specified column to null
+ * @param columnName Name of the column
+ * @throws IllegalArgumentException if the column doesn't exist or cannot be set to null
+ * @throws IllegalStateException if the row was already applied
+ */
+ public void setNull(String columnName) {
+ setNull(this.schema.getColumn(columnName));
+ }
+
+ private void setNull(ColumnSchema column) {
+ assert nullsBitSet != null;
+ checkNotFrozen();
+ checkColumnExists(column);
+ if (!column.isNullable()) {
+ throw new IllegalArgumentException(column.getName() + " cannot be set to null");
+ }
+ int idx = schema.getColumns().indexOf(column);
+ columnsBitSet.set(idx);
+ nullsBitSet.set(idx);
+ }
+
+ /**
+ * Verifies if the column exists and belongs to one of the specified types
+ * It also does some internal accounting
+ * @param column column the user wants to set
+ * @param types types we expect
+ * @throws IllegalArgumentException if the column or type was invalid
+ * @throws IllegalStateException if the row was already applied
+ */
+ private void checkColumn(ColumnSchema column, Type... types) {
+ checkNotFrozen();
+ checkColumnExists(column);
+ for(Type type : types) {
+ if (column.getType().equals(type)) return;
+ }
+ throw new IllegalArgumentException(String.format("%s isn't %s, it's %s", column.getName(),
+ Arrays.toString(types), column.getType().getName()));
+ }
+
+ /**
+ * @param column column the user wants to set
+ * @throws IllegalArgumentException if the column doesn't exist
+ */
+ private void checkColumnExists(ColumnSchema column) {
+ if (column == null)
+ throw new IllegalArgumentException("Column name isn't present in the table's schema");
+ }
+
+ /**
+ * @throws IllegalStateException if the row was already applied
+ */
+ private void checkNotFrozen() {
+ if (frozen) {
+ throw new IllegalStateException("This row was already applied and cannot be modified.");
+ }
+ }
+
+ /**
+ * Sets the column bit set for the column index, and returns the column's offset.
+ * @param columnIndex the index of the column to get the position for and mark as set
+ * @return the offset in rowAlloc for the column
+ */
+ private int getPositionInRowAllocAndSetBitSet(int columnIndex) {
+ columnsBitSet.set(columnIndex);
+ return schema.getColumnOffset(columnIndex);
+ }
+
+ /**
+ * Tells if the specified column was set by the user
+ * @param column column's index in the schema
+ * @return true if it was set, else false
+ */
+ boolean isSet(int column) {
+ return this.columnsBitSet.get(column);
+ }
+
+ /**
+ * Tells if the specified column was set to null by the user
+ * @param column column's index in the schema
+ * @return true if it was set, else false
+ */
+ boolean isSetToNull(int column) {
+ if (this.nullsBitSet == null) {
+ return false;
+ }
+ return this.nullsBitSet.get(column);
+ }
+
+ /**
+ * Returns the encoded primary key of the row.
+ * @return a byte array containing an encoded primary key
+ */
+ public byte[] encodePrimaryKey() {
+ return new KeyEncoder().encodePrimaryKey(this);
+ }
+
+ /**
+ * Transforms the row key into a string representation where each column is in the format:
+ * "type col_name=value".
+ * @return a string representation of the operation's row key
+ */
+ public String stringifyRowKey() {
+ int numRowKeys = schema.getPrimaryKeyColumnCount();
+ StringBuilder sb = new StringBuilder();
+ sb.append("(");
+ for (int i = 0; i < numRowKeys; i++) {
+ if (i > 0) {
+ sb.append(", ");
+ }
+
+ ColumnSchema col = schema.getColumnByIndex(i);
+ assert !col.isNullable();
+ Preconditions.checkState(columnsBitSet.get(i),
+ "Full row key not specified, missing at least col: " + col.getName());
+ Type type = col.getType();
+ sb.append(type.getName());
+ sb.append(" ");
+ sb.append(col.getName());
+ sb.append("=");
+
+ if (type == Type.STRING || type == Type.BINARY) {
+ ByteBuffer value = getVarLengthData().get(i).duplicate();
+ value.reset(); // Make sure we start at the beginning.
+ byte[] data = new byte[value.limit()];
+ value.get(data);
+ if (type == Type.STRING) {
+ sb.append(Bytes.getString(data));
+ } else {
+ sb.append(Bytes.pretty(data));
+ }
+ } else {
+ switch (type) {
+ case INT8:
+ sb.append(Bytes.getByte(rowAlloc, schema.getColumnOffset(i)));
+ break;
+ case INT16:
+ sb.append(Bytes.getShort(rowAlloc, schema.getColumnOffset(i)));
+ break;
+ case INT32:
+ sb.append(Bytes.getInt(rowAlloc, schema.getColumnOffset(i)));
+ break;
+ case INT64:
+ sb.append(Bytes.getLong(rowAlloc, schema.getColumnOffset(i)));
+ break;
+ case TIMESTAMP:
+ sb.append(Bytes.getLong(rowAlloc, schema.getColumnOffset(i)));
+ break;
+ default:
+ throw new IllegalArgumentException(String.format(
+ "The column type %s is not a valid key component type", type));
+ }
+ }
+ }
+ sb.append(")");
+
+ return sb.toString();
+ }
+
+ /**
+ * Get the schema used for this row.
+ * @return a schema that came from KuduTable
+ */
+ Schema getSchema() {
+ return schema;
+ }
+
+ /**
+ * Get the list variable length data cells that were added to this row.
+ * @return a list of binary data, may be empty
+ */
+ List<ByteBuffer> getVarLengthData() {
+ return varLengthData;
+ }
+
+ /**
+ * Get the byte array that contains all the data added to this partial row. Variable length data
+ * is contained separately, see {@link #getVarLengthData()}. In their place you'll find their
+ * index in that list and their size.
+ * @return a byte array containing the data for this row, except strings
+ */
+ byte[] getRowAlloc() {
+ return rowAlloc;
+ }
+
+ /**
+ * Get the bit set that indicates which columns were set.
+ * @return a bit set for columns with data
+ */
+ BitSet getColumnsBitSet() {
+ return columnsBitSet;
+ }
+
+ /**
+ * Get the bit set for the columns that were specifically set to null
+ * @return a bit set for null columns
+ */
+ BitSet getNullsBitSet() {
+ return nullsBitSet;
+ }
+
+ /**
+ * Prevents this PartialRow from being modified again. Can be called multiple times.
+ */
+ void freeze() {
+ this.frozen = true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/main/java/org/apache/kudu/client/Partition.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/Partition.java b/java/kudu-client/src/main/java/org/apache/kudu/client/Partition.java
new file mode 100644
index 0000000..bdc089b
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/Partition.java
@@ -0,0 +1,182 @@
+// 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.Objects;
+import org.kududb.annotations.InterfaceAudience;
+import org.kududb.annotations.InterfaceStability;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * A Partition describes the set of rows that a Tablet is responsible for
+ * serving. Each tablet is assigned a single Partition.<p>
+ *
+ * Partitions consist primarily of a start and end partition key. Every row with
+ * a partition key that falls in a Tablet's Partition will be served by that
+ * tablet.<p>
+ *
+ * In addition to the start and end partition keys, a Partition holds metadata
+ * to determine if a scan can prune, or skip, a partition based on the scan's
+ * start and end primary keys, and predicates.
+ *
+ * This class is new, and not considered stable or suitable for public use.
+ */
+@InterfaceAudience.LimitedPrivate("Impala")
+@InterfaceStability.Unstable
+public class Partition implements Comparable<Partition> {
+ final byte[] partitionKeyStart;
+ final byte[] partitionKeyEnd;
+
+ final byte[] rangeKeyStart;
+ final byte[] rangeKeyEnd;
+
+ final List<Integer> hashBuckets;
+
+ /**
+ * Size of an encoded hash bucket component in a partition key.
+ */
+ private static final int ENCODED_BUCKET_SIZE = 4;
+
+ /**
+ * Creates a new partition with the provided start and end keys, and hash buckets.
+ * @param partitionKeyStart the start partition key
+ * @param partitionKeyEnd the end partition key
+ * @param hashBuckets the partition hash buckets
+ */
+ Partition(byte[] partitionKeyStart,
+ byte[] partitionKeyEnd,
+ List<Integer> hashBuckets) {
+ this.partitionKeyStart = partitionKeyStart;
+ this.partitionKeyEnd = partitionKeyEnd;
+ this.hashBuckets = hashBuckets;
+ this.rangeKeyStart = rangeKey(partitionKeyStart, hashBuckets.size());
+ this.rangeKeyEnd = rangeKey(partitionKeyEnd, hashBuckets.size());
+ }
+
+ /**
+ * Gets the start partition key.
+ * @return the start partition key
+ */
+ public byte[] getPartitionKeyStart() {
+ return partitionKeyStart;
+ }
+
+ /**
+ * Gets the end partition key.
+ * @return the end partition key
+ */
+ public byte[] getPartitionKeyEnd() {
+ return partitionKeyEnd;
+ }
+
+ /**
+ * Gets the start range key.
+ * @return the start range key
+ */
+ public byte[] getRangeKeyStart() {
+ return rangeKeyStart;
+ }
+
+ /**
+ * Gets the end range key.
+ * @return the end range key
+ */
+ public byte[] getRangeKeyEnd() {
+ return rangeKeyEnd;
+ }
+
+ /**
+ * Gets the partition hash buckets.
+ * @return the partition hash buckets
+ */
+ public List<Integer> getHashBuckets() {
+ return hashBuckets;
+ }
+
+ /**
+ * @return true if the partition is the absolute end partition
+ */
+ public boolean isEndPartition() {
+ return partitionKeyEnd.length == 0;
+ }
+
+ /**
+ * Equality only holds for partitions from the same table. Partition equality only takes into
+ * account the partition keys, since there is a 1 to 1 correspondence between partition keys and
+ * the hash buckets and range keys.
+ *
+ * @return the hash code
+ */
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ Partition partition = (Partition) o;
+ return Arrays.equals(partitionKeyStart, partition.partitionKeyStart)
+ && Arrays.equals(partitionKeyEnd, partition.partitionKeyEnd);
+ }
+
+ /**
+ * The hash code only takes into account the partition keys, since there is a 1 to 1
+ * correspondence between partition keys and the hash buckets and range keys.
+ *
+ * @return the hash code
+ */
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(Arrays.hashCode(partitionKeyStart), Arrays.hashCode(partitionKeyEnd));
+ }
+
+ /**
+ * Partition comparison is only reasonable when comparing partitions from the same table, and
+ * since Kudu does not yet allow partition splitting, no two distinct partitions can have the
+ * same start partition key. Accordingly, partitions are compared strictly by the start partition
+ * key.
+ *
+ * @param other the other partition of the same table
+ * @return the comparison of the partitions
+ */
+ @Override
+ public int compareTo(Partition other) {
+ return Bytes.memcmp(this.partitionKeyStart, other.partitionKeyStart);
+ }
+
+ /**
+ * Returns the range key portion of a partition key given the number of buckets in the partition
+ * schema.
+ * @param partitionKey the partition key containing the range key
+ * @param numHashBuckets the number of hash bucket components of the table
+ * @return the range key
+ */
+ private static byte[] rangeKey(byte[] partitionKey, int numHashBuckets) {
+ int bucketsLen = numHashBuckets * ENCODED_BUCKET_SIZE;
+ if (partitionKey.length > bucketsLen) {
+ return Arrays.copyOfRange(partitionKey, bucketsLen, partitionKey.length);
+ } else {
+ return AsyncKuduClient.EMPTY_ARRAY;
+ }
+ }
+
+ @Override
+ public String toString() {
+ return String.format("[%s, %s)",
+ partitionKeyStart.length == 0 ? "<start>" : Bytes.hex(partitionKeyStart),
+ partitionKeyEnd.length == 0 ? "<end>" : Bytes.hex(partitionKeyEnd));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/main/java/org/apache/kudu/client/PartitionSchema.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/PartitionSchema.java b/java/kudu-client/src/main/java/org/apache/kudu/client/PartitionSchema.java
new file mode 100644
index 0000000..fdee32e
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/PartitionSchema.java
@@ -0,0 +1,142 @@
+// 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.kududb.Schema;
+import org.kududb.annotations.InterfaceAudience;
+import org.kududb.annotations.InterfaceStability;
+
+import java.util.List;
+
+/**
+ * A partition schema describes how the rows of a table are distributed among
+ * tablets.
+ *
+ * Primarily, a table's partition schema is responsible for translating the
+ * primary key column values of a row into a partition key that can be used to
+ * find the tablet containing the key.
+ *
+ * The partition schema is made up of zero or more hash bucket components,
+ * followed by a single range component.
+ *
+ * Each hash bucket component includes one or more columns from the primary key
+ * column set, with the restriction that an individual primary key column may
+ * only be included in a single hash component.
+ *
+ * This class is new, and not considered stable or suitable for public use.
+ */
+@InterfaceAudience.LimitedPrivate("Impala")
+@InterfaceStability.Unstable
+public class PartitionSchema {
+
+ private final RangeSchema rangeSchema;
+ private final List<HashBucketSchema> hashBucketSchemas;
+ private final boolean isSimple;
+
+ /**
+ * Creates a new partition schema from the range and hash bucket schemas.
+ *
+ * @param rangeSchema the range schema
+ * @param hashBucketSchemas the hash bucket schemas
+ * @param schema the table schema
+ */
+ PartitionSchema(RangeSchema rangeSchema,
+ List<HashBucketSchema> hashBucketSchemas,
+ Schema schema) {
+ this.rangeSchema = rangeSchema;
+ this.hashBucketSchemas = hashBucketSchemas;
+
+ boolean isSimple = hashBucketSchemas.isEmpty()
+ && rangeSchema.columns.size() == schema.getPrimaryKeyColumnCount();
+ if (isSimple) {
+ int i = 0;
+ for (Integer id : rangeSchema.columns) {
+ if (schema.getColumnIndex(id) != i++) {
+ isSimple = false;
+ break;
+ }
+ }
+ }
+ this.isSimple = isSimple;
+ }
+
+ /**
+ * Returns the encoded partition key of the row.
+ * @return a byte array containing the encoded partition key of the row
+ */
+ public byte[] encodePartitionKey(PartialRow row) {
+ return new KeyEncoder().encodePartitionKey(row, this);
+ }
+
+ public RangeSchema getRangeSchema() {
+ return rangeSchema;
+ }
+
+ public List<HashBucketSchema> getHashBucketSchemas() {
+ return hashBucketSchemas;
+ }
+
+ /**
+ * Returns true if the partition schema if the partition schema does not include any hash
+ * components, and the range columns match the table's primary key columns.
+ *
+ * @return whether the partition schema is the default simple range partitioning.
+ */
+ boolean isSimpleRangePartitioning() {
+ return isSimple;
+ }
+
+ public static class RangeSchema {
+ private final List<Integer> columns;
+
+ RangeSchema(List<Integer> columns) {
+ this.columns = columns;
+ }
+
+ public List<Integer> getColumns() {
+ return columns;
+ }
+ }
+
+ public static class HashBucketSchema {
+ private final List<Integer> columnIds;
+ private int numBuckets;
+ private int seed;
+
+ HashBucketSchema(List<Integer> columnIds, int numBuckets, int seed) {
+ this.columnIds = columnIds;
+ this.numBuckets = numBuckets;
+ this.seed = seed;
+ }
+
+ /**
+ * Gets the column IDs of the columns in the hash partition.
+ * @return the column IDs of the columns in the has partition
+ */
+ public List<Integer> getColumnIds() {
+ return columnIds;
+ }
+
+ public int getNumBuckets() {
+ return numBuckets;
+ }
+
+ public int getSeed() {
+ return seed;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/main/java/org/apache/kudu/client/PleaseThrottleException.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/PleaseThrottleException.java b/java/kudu-client/src/main/java/org/apache/kudu/client/PleaseThrottleException.java
new file mode 100644
index 0000000..3ca98e2
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/PleaseThrottleException.java
@@ -0,0 +1,105 @@
+/*
+ * Copyright (C) 2010-2012 The Async HBase Authors. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ * - Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ * - Redistributions in binary form must reproduce the above copyright notice,
+ * this list of conditions and the following disclaimer in the documentation
+ * and/or other materials provided with the distribution.
+ * - Neither the name of the StumbleUpon nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+package org.kududb.client;
+
+import com.stumbleupon.async.Deferred;
+import org.kududb.annotations.InterfaceAudience;
+import org.kududb.annotations.InterfaceStability;
+
+/**
+ * This exception notifies the application to throttle its use of Kudu.
+ * <p>
+ * Since all APIs of {@link AsyncKuduSession} are asynchronous and non-blocking,
+ * it's possible that the application would produce RPCs at a rate higher
+ * than Kudu is able to handle. When this happens, {@link AsyncKuduSession}
+ * will typically do some buffering up to a certain point beyond which RPCs
+ * will fail-fast with this exception, to prevent the application from
+ * running itself out of memory.
+ * <p>
+ * This exception is expected to be handled by having the application
+ * throttle or pause itself for a short period of time before retrying the
+ * RPC that failed with this exception as well as before sending other RPCs.
+ * The reason this exception inherits from {@link NonRecoverableException}
+ * instead of {@link RecoverableException} is that the usual course of action
+ * when handling a {@link RecoverableException} is to retry right away, which
+ * would defeat the whole purpose of this exception. Here, we want the
+ * application to <b>retry after a reasonable delay</b> as well as <b>throttle
+ * the pace of creation of new RPCs</b>. What constitutes a "reasonable
+ * delay" depends on the nature of RPCs and rate at which they're produced.
+ * <p>
+ * One effective strategy to handle this exception is to set a flag to true
+ * when this exception is first emitted that causes the application to pause
+ * or throttle its use of Kudu. Then you can retry the RPC that failed
+ * (which is accessible through {@link #getFailedRpc}) and add a callback to
+ * it in order to unset the flag once the RPC completes successfully.
+ * Note that low-throughput applications will typically rarely (if ever)
+ * hit this exception, so they don't need complex throttling logic.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+@SuppressWarnings("serial")
+public final class PleaseThrottleException extends RecoverableException
+ implements HasFailedRpcException {
+
+ /** The RPC that was failed with this exception. */
+ private final Operation rpc;
+
+ /** A deferred one can wait on before retrying the failed RPC. */
+ private final Deferred deferred;
+
+ /**
+ * Constructor.
+ * @param status status object containing the reason for the exception
+ * @param cause The exception that requires the application to throttle
+ * itself (can be {@code null})
+ * @param rpc The RPC that was made to fail with this exception
+ * @param deferred A deferred one can wait on before retrying the failed RPC
+ */
+ PleaseThrottleException(Status status,
+ KuduException cause,
+ Operation rpc,
+ Deferred deferred) {
+ super(status, cause);
+ this.rpc = rpc;
+ this.deferred = deferred;
+ }
+
+ /**
+ * The RPC that was made to fail with this exception.
+ */
+ public Operation getFailedRpc() {
+ return rpc;
+ }
+
+ /**
+ * Returns a deferred one can wait on before retrying the failed RPC.
+ * @since 1.3
+ */
+ public Deferred getDeferred() {
+ return deferred;
+ }
+
+}