You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by kt...@apache.org on 2017/03/20 14:49:04 UTC
[8/9] accumulo git commit: ACCUMULO-4501 ACCUMULO-96 Added
Summarization
http://git-wip-us.apache.org/repos/asf/accumulo/blob/94cdcc4d/core/src/main/java/org/apache/accumulo/core/client/summary/Summarizer.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/summary/Summarizer.java b/core/src/main/java/org/apache/accumulo/core/client/summary/Summarizer.java
new file mode 100644
index 0000000..fdb194b
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/client/summary/Summarizer.java
@@ -0,0 +1,227 @@
+/*
+ * 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.apache.accumulo.core.client.summary;
+
+import java.util.Map;
+
+import org.apache.accumulo.core.client.admin.TableOperations;
+import org.apache.accumulo.core.client.rfile.RFile;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+//checkstyle and the formatter are in conflict, so turn off the formatter
+//@formatter:off
+/**
+ * <p>
+ * Instances of this interface can be configured for Accumulo tables. When Accumulo compacts files, it will use this Factory to create {@link Collector} and
+ * {@link Combiner} objects to generate summary information about the data in the file.
+ *
+ * <p>
+ * In order to merge summary information from multiple files, Accumulo will use this factory to create a {@link Combiner} object.
+ *
+ * <p>
+ * Below is an example of a very simple summarizer that will compute the number of deletes, total number of keys, min timestamp and max timestamp.
+ *
+ * <pre>
+ * <code>
+ * public class BasicSummarizer implements Summarizer {
+ *
+ * public static final String DELETES_STAT = "deletes";
+ * public static final String MIN_STAMP_STAT = "minStamp";
+ * public static final String MAX_STAMP_STAT = "maxStamp";
+ * public static final String TOTAL_STAT = "total";
+ *
+ * @Override
+ * public Collector collector(SummarizerConfiguration sc) {
+ * return new Collector() {
+ *
+ * private long minStamp = Long.MAX_VALUE;
+ * private long maxStamp = Long.MIN_VALUE;
+ * private long deletes = 0;
+ * private long total = 0;
+ *
+ * @Override
+ * public void accept(Key k, Value v) {
+ * if (k.getTimestamp() < minStamp) {
+ * minStamp = k.getTimestamp();
+ * }
+ *
+ * if (k.getTimestamp() > maxStamp) {
+ * maxStamp = k.getTimestamp();
+ * }
+ *
+ * if (k.isDeleted()) {
+ * deletes++;
+ * }
+ *
+ * total++;
+ * }
+ *
+ * @Override
+ * public void summarize(StatisticConsumer sc) {
+ * sc.accept(MIN_STAMP_STAT, minStamp);
+ * sc.accept(MAX_STAMP_STAT, maxStamp);
+ * sc.accept(DELETES_STAT, deletes);
+ * sc.accept(TOTAL_STAT, total);
+ * }
+ * };
+ * }
+ *
+ * @Override
+ * public Combiner combiner(SummarizerConfiguration sc) {
+ * return (stats1, stats2) -> {
+ * stats1.merge(DELETES_STAT, stats2.get(DELETES_STAT), Long::sum);
+ * stats1.merge(TOTAL_STAT, stats2.get(TOTAL_STAT), Long::sum);
+ * stats1.merge(MIN_STAMP_STAT, stats2.get(MIN_STAMP_STAT), Long::min);
+ * stats1.merge(MAX_STAMP_STAT, stats2.get(MAX_STAMP_STAT), Long::max);
+ * };
+ * }
+ * }
+ * </code>
+ * </pre>
+ *
+ * <p>
+ * Below is an example summarizer that counts the log of the value length.
+ *
+ * <pre>
+ * <code>
+ * public class ValueLogLengthSummarizer implements Summarizer {
+ *
+ * @Override
+ * public Collector collector(SummarizerConfiguration sc) {
+ *
+ * return new Collector(){
+ *
+ * long[] counts = new long[32];
+ *
+ * @Override
+ * public void accept(Key k, Value v) {
+ * int idx;
+ * if(v.getSize() == 0)
+ * idx = 0;
+ * else
+ * idx = IntMath.log2(v.getSize(), RoundingMode.UP); //IntMath is from Guava
+ *
+ * counts[idx]++;
+ * }
+ *
+ * @Override
+ * public void summarize(StatisticConsumer sc) {
+ * for (int i = 0; i < counts.length; i++) {
+ * if(counts[i] > 0) {
+ * sc.accept(""+(1<<i), counts[i]);
+ * }
+ * }
+ * }
+ * };
+ * }
+ *
+ * @Override
+ * public Combiner combiner(SummarizerConfiguration sc) {
+ * return (m1, m2) -> m2.forEach((k,v) -> m1.merge(k, v, Long::sum));
+ * }
+ * }
+ * </code>
+ * </pre>
+ *
+ * <p>
+ * The reason a Summarizer is a factory for a Collector and Combiner is to make it very clear in the API that Accumulo uses them independently at different
+ * times. Therefore its not advisable to share internal state between the Collector and Combiner. The example implementation shows that the Collectors design
+ * allows for very efficient collection of specialized summary information. Creating {@link String} + {@link Long} pairs is deferred until the summarize method
+ * is called.
+ *
+ * <p>
+ * Summary data can be used by Compaction Strategies to decide which files to compact.
+ *
+ * <p>
+ * Summary data is persisted, so ideally the same summarizer class with the same options should always produce the same results. If you need to change the behavior
+ * of a summarizer, then consider doing this by adding a new option. If the same summarizer is configured twice with different options, then Accumulo will store and
+ * merge each one separately. This can allow old and new behavior to coexists simultaneously.
+ *
+ * @since 2.0.0
+ *
+ * @see TableOperations#summaries(String)
+ * @see TableOperations#addSummarizers(String, SummarizerConfiguration...)
+ * @see TableOperations#listSummarizers(String)
+ * @see TableOperations#removeSummarizers(String, java.util.function.Predicate)
+ * @see RFile#summaries()
+ * @see SummarizerConfiguration
+ */
+ //@formatter:on
+public interface Summarizer {
+
+ public static interface StatisticConsumer {
+ public void accept(String statistic, long value);
+ }
+
+ /**
+ * When Accumulo calls methods in this interface, it will call {@link #accept(Key, Value)} zero or more times and then call
+ * {@link #summarize(Summarizer.StatisticConsumer)} once. After calling {@link #summarize(Summarizer.StatisticConsumer)}, it will not use the collector again.
+ *
+ * @since 2.0.0
+ */
+ public static interface Collector {
+ /**
+ * During compactions, Accumulo passes each Key Value written to the file to this method.
+ */
+ void accept(Key k, Value v);
+
+ /**
+ * After Accumulo has written some Key Values, it will call this method to generate some statistics about what was previously passed to
+ * {@link #accept(Key, Value)}.
+ *
+ * <p>
+ * In order for summary data to be useful for decision making about data, it needs to be quickly accessible. In order to be quickly accessible, it needs to
+ * fit in the tablet server cache as described in {@link TableOperations#summaries(String)} and the compaction strategy documentation. Therefore its
+ * advisable to generate small summaries. If the summary data generated is too large it will not be stored. The maximum summary size is set using the per
+ * table property {@code table.file.summary.maxSize}. The number of files that exceeded the summary size is reported by
+ * {@link Summary.FileStatistics#getLarge()}.
+ *
+ * @param sc
+ * Emit statistics to this Object.
+ */
+ public void summarize(StatisticConsumer sc);
+ }
+
+ /**
+ * A Combiner is used to merge statistics emitted from {@link Collector#summarize(StatisticConsumer)} and from previous invocations of itself.
+ *
+ * @since 2.0.0
+ */
+ public static interface Combiner {
+ /**
+ * This method should merge the statistics in the second map into the first map. Both maps may have statistics produced by a {@link Collector} or previous
+ * calls to this method.
+ *
+ * <p>
+ * If first map is too large after this call, then it may not be stored. See the comment on {@link Collector#summarize(StatisticConsumer)}
+ */
+ public void merge(Map<String,Long> statistics1, Map<String,Long> statistics2);
+ }
+
+ /**
+ * Factory method that creates a {@link Collector} based on configuration. Each {@link Collector} created by this method should be independent and have its
+ * own internal state. Accumulo uses a Collector to generate summary statistics about a sequence of key values written to a file.
+ */
+ public Collector collector(SummarizerConfiguration sc);
+
+ /**
+ * Factory method that creates a {@link Combiner}. Accumulo will only use the created Combiner to merge data from {@link Collector}s created using the same
+ * {@link SummarizerConfiguration}.
+ */
+ public Combiner combiner(SummarizerConfiguration sc);
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/94cdcc4d/core/src/main/java/org/apache/accumulo/core/client/summary/SummarizerConfiguration.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/summary/SummarizerConfiguration.java b/core/src/main/java/org/apache/accumulo/core/client/summary/SummarizerConfiguration.java
new file mode 100644
index 0000000..ec98695
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/client/summary/SummarizerConfiguration.java
@@ -0,0 +1,285 @@
+/*
+ * 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.apache.accumulo.core.client.summary;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.accumulo.core.summary.SummarizerConfigurationUtil;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.hash.Hasher;
+import com.google.common.hash.Hashing;
+
+/**
+ * This class encapsulates the configuration needed to instantiate a {@link Summarizer}. It also provides methods and documentation for setting the table
+ * properties that configure a Summarizer.
+ *
+ * @since 2.0.0
+ */
+public class SummarizerConfiguration {
+
+ private final String className;
+ private final Map<String,String> options;
+ private int hashCode = 0;
+ private final String configId;
+
+ private SummarizerConfiguration(String className, String configId, Map<String,String> options) {
+ this.className = className;
+ this.options = ImmutableMap.copyOf(options);
+
+ if (configId == null) {
+ ArrayList<String> keys = new ArrayList<>(this.options.keySet());
+ Collections.sort(keys);
+ Hasher hasher = Hashing.murmur3_32().newHasher();
+ hasher.putString(className);
+ for (String key : keys) {
+ hasher.putString(key);
+ hasher.putString(options.get(key));
+ }
+
+ this.configId = hasher.hash().toString();
+ } else {
+ this.configId = configId;
+ }
+ }
+
+ /**
+ * @return the name of a class that implements @link {@link Summarizer}.
+ */
+ public String getClassName() {
+ return className;
+ }
+
+ /**
+ * @return custom options for a {link @Summarizer}
+ */
+ public Map<String,String> getOptions() {
+ return options;
+ }
+
+ /**
+ * The propertyId is used to when creating table properties for a summarizer. Its not used for equality or hashCode for this class.
+ */
+ public String getPropertyId() {
+ return configId;
+ }
+
+ @Override
+ public String toString() {
+ return className + " " + configId + " " + options;
+ }
+
+ /**
+ * Compares the classname and options to determine equality.
+ */
+ @Override
+ public boolean equals(Object o) {
+ if (o instanceof SummarizerConfiguration) {
+ SummarizerConfiguration osc = (SummarizerConfiguration) o;
+ return className.equals(osc.className) && options.equals(osc.options);
+ }
+
+ return false;
+ }
+
+ /**
+ * Hashes the classname and options to create a hashcode.
+ */
+ @Override
+ public int hashCode() {
+ if (hashCode == 0) {
+ hashCode = 31 * options.hashCode() + className.hashCode();
+ }
+ return hashCode;
+ }
+
+ /**
+ * Converts this configuration to Accumulo per table properties. The returned map has the following key values. The {@code <configId>} below is from
+ * {@link #getPropertyId()}. The {@code <optionKey>} and {@code <optionValue>} below are derived from the key values of {@link #getOptions()}.
+ *
+ * <pre>
+ * {@code
+ * table.summarizer.<configId>=<classname>
+ * table.summarizer.<configId>.opt.<optionKey1>=<optionValue1>
+ * table.summarizer.<configId>.opt.<optionKey2>=<optionValue2>
+ * .
+ * .
+ * .
+ * table.summarizer.<configId>.opt.<optionKeyN>=<optionValueN>
+ * }
+ * </pre>
+ */
+ public Map<String,String> toTableProperties() {
+ return SummarizerConfigurationUtil.toTablePropertiesMap(Collections.singletonList(this));
+ }
+
+ /**
+ * Encodes each configuration in the same way as {@link #toTableProperties()}.
+ *
+ * @throws IllegalArgumentException
+ * when there are duplicate values for {@link #getPropertyId()}
+ */
+ public static Map<String,String> toTableProperties(SummarizerConfiguration... configurations) {
+ return SummarizerConfigurationUtil.toTablePropertiesMap(Arrays.asList(configurations));
+ }
+
+ /**
+ * Encodes each configuration in the same way as {@link #toTableProperties()}.
+ *
+ * @throws IllegalArgumentException
+ * when there are duplicate values for {@link #getPropertyId()}
+ */
+ public static Map<String,String> toTableProperties(Collection<SummarizerConfiguration> configurations) {
+ return SummarizerConfigurationUtil.toTablePropertiesMap(new ArrayList<SummarizerConfiguration>(configurations));
+ }
+
+ /**
+ * Decodes table properties with the prefix {@code table.summarizer} into {@link SummarizerConfiguration} objects. Table properties with prefixes other than
+ * {@code table.summarizer} are ignored.
+ */
+ public static Collection<SummarizerConfiguration> fromTableProperties(Map<String,String> props) {
+ return fromTableProperties(props.entrySet());
+ }
+
+ /**
+ * @see #fromTableProperties(Map)
+ */
+ public static Collection<SummarizerConfiguration> fromTableProperties(Iterable<Entry<String,String>> props) {
+ return SummarizerConfigurationUtil.getSummarizerConfigs(props);
+ }
+
+ public static class Builder {
+ private String className;
+ private ImmutableMap.Builder<String,String> imBuilder;
+ private String configId = null;
+
+ private Builder(String className) {
+ this.className = className;
+ this.imBuilder = ImmutableMap.builder();
+ }
+
+ /**
+ * Sets the id used when generating table properties. Setting this is optional. If not set, an id is generated using hashing that will likely be unique.
+ *
+ * @param propId
+ * This id is used when converting a {@link SummarizerConfiguration} to table properties. Since tables can have multiple summarizers, make sure its
+ * unique.
+ *
+ * @see SummarizerConfiguration#toTableProperties()
+ */
+ public Builder setPropertyId(String propId) {
+ Preconditions.checkArgument(propId.matches("\\w+"), "Config Id %s is not alphanum", propId);
+ this.configId = propId;
+ return this;
+ }
+
+ /**
+ * Adds an option that Summarizers can use when constructing Collectors and Combiners.
+ *
+ * @return this
+ *
+ * @see SummarizerConfiguration#getOptions()
+ */
+ public Builder addOption(String key, String value) {
+ Preconditions.checkArgument(key.matches("\\w+"), "Option Id %s is not alphanum", key);
+ imBuilder.put(key, value);
+ return this;
+ }
+
+ /**
+ * Adds an option that Summarizers can use when constructing Collectors and Combiners.
+ *
+ * @return this
+ *
+ * @see SummarizerConfiguration#getOptions()
+ */
+ public Builder addOption(String key, long value) {
+ return addOption(key, Long.toString(value));
+ }
+
+ /**
+ * Convenience method for adding multiple options. The following
+ *
+ * <pre>
+ * {@code builder.addOptions("opt1","val1","opt2","val2","opt3","val3")}
+ * </pre>
+ *
+ * <p>
+ * is equivalent to
+ *
+ * <pre>
+ * {@code
+ * builder.addOption("opt1","val1");
+ * builder.addOption("opt2","val2");
+ * builder.addOption("opt3","val3");
+ * }
+ * </pre>
+ *
+ * @param keyValuePairs
+ * This array must have an even and positive number of elements.
+ * @return this
+ * @see SummarizerConfiguration#getOptions()
+ */
+ public Builder addOptions(String... keyValuePairs) {
+ Preconditions.checkArgument(keyValuePairs.length % 2 == 0 && keyValuePairs.length > 0, "Require an even, positive number of arguments, got %s",
+ keyValuePairs.length);
+ for (int i = 0; i < keyValuePairs.length; i += 2) {
+ addOption(keyValuePairs[i], keyValuePairs[i + 1]);
+ }
+ return this;
+ }
+
+ /**
+ * @param options
+ * Each entry in the map is passed to {@link #addOption(String, String)}
+ * @return this
+ *
+ * @see SummarizerConfiguration#getOptions()
+ */
+ public Builder addOptions(Map<String,String> options) {
+ options.entrySet().forEach(e -> addOption(e.getKey(), e.getValue()));
+ return this;
+ }
+
+ public SummarizerConfiguration build() {
+ return new SummarizerConfiguration(className, configId, imBuilder.build());
+ }
+ }
+
+ /**
+ * Call this method to initiate a chain of fluent method calls to a create an immutable {@link SummarizerConfiguration}
+ *
+ * @param className
+ * The fully qualified name of a class that implements {@link Summarizer}.
+ */
+ public static Builder builder(String className) {
+ return new Builder(className);
+ }
+
+ /**
+ * @see #builder(String)
+ */
+ public static Builder builder(Class<? extends Summarizer> clazz) {
+ return new Builder(clazz.getName());
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/94cdcc4d/core/src/main/java/org/apache/accumulo/core/client/summary/Summary.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/summary/Summary.java b/core/src/main/java/org/apache/accumulo/core/client/summary/Summary.java
new file mode 100644
index 0000000..8a6a9aa
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/client/summary/Summary.java
@@ -0,0 +1,145 @@
+/*
+ * 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.apache.accumulo.core.client.summary;
+
+import java.util.Map;
+
+import com.google.common.collect.ImmutableMap;
+
+/**
+ * This class encapsulates summary statistics, information about how those statistics were generated, and information about files the statistics were obtained
+ * from.
+ *
+ * @see Summarizer
+ * @since 2.0.0
+ */
+public class Summary {
+
+ /**
+ * @since 2.0.0
+ */
+ public static class FileStatistics {
+ private final long total;
+ private final long missing;
+ private final long extra;
+ private final long large;
+ private final long deleted;
+
+ private FileStatistics(long total, long missing, long extra, long large, long deleted) {
+ this.total = total;
+ this.missing = missing;
+ this.extra = extra;
+ this.large = large;
+ this.deleted = deleted;
+ }
+
+ /**
+ * @return The total number of files from which summary information was obtained.
+ */
+ public long getTotal() {
+ return total;
+ }
+
+ /**
+ * @return The number of files that did not contain the requested summary information. When this is non-zero, it means that summary counts may be
+ * incomplete. In the Accumulo shell, the compact command has a -{@code -sf-no-summary} option to compact files missing summary information. The
+ * compaction will create the summary information. This could be done over a range of the table to avoid doing the entire table at once.
+ */
+ public long getMissing() {
+ return missing;
+ }
+
+ /**
+ * @return The number of files that had summary information outside of a tablet or query range boundaries. When this is non-zero, it means that summary
+ * counts may be artificially inflated or contain extraneous information. In the Accumulo shell, the compact command has a -{@code -sf-extra-summary}
+ * option to compact files with extra summary information.
+ */
+ public long getExtra() {
+ return extra;
+ }
+
+ /**
+ * @return The number of files that an attempt was made to generate summaries, but the summarizer generated a summary that was larger than the configured
+ * maximum. For these files no summary statistics are stored. Only the fact that summarization was attempted and failed is stored.
+ * @see Summarizer.Collector#summarize(org.apache.accumulo.core.client.summary.Summarizer.StatisticConsumer)
+ */
+ public long getLarge() {
+ return large;
+ }
+
+ /**
+ * @return The number of files that were deleted after the summary retrieval operations started. This is a rare race condition where a compaction causes a
+ * file to be deleted while retrieving summaries. When this happens, the file that replaced the deleted file can not be used because it may contain
+ * duplication summary information for other files. Avoiding this race condition would be expensive, so reporting it was chosen. If this condition
+ * must be avoided, then compactions must be stopped. Compactions could be stopped on a cloned table to avoid this.
+ */
+ public long getDeleted() {
+ return deleted;
+ }
+
+ /**
+ * @return The total number of files that had some kind of issue which would cause summary statistics to be inaccurate. This is the sum of
+ * {@link #getMissing()}, {@link #getExtra()}, {{@link #getLarge()}, and {@link #getDeleted()}.
+ */
+ public long getInaccurate() {
+ return getMissing() + getExtra() + getLarge() + getDeleted();
+ }
+
+ @Override
+ public String toString() {
+ return String.format("[total:%,d, missing:%,d, extra:%,d, large:%,d, deleted:%,d]", total, missing, extra, large, deleted);
+ }
+ }
+
+ private final ImmutableMap<String,Long> statistics;
+ private final SummarizerConfiguration config;
+ private final FileStatistics fileStats;
+
+ public Summary(Map<String,Long> summary, SummarizerConfiguration config, long totalFiles, long filesMissingSummary, long filesWithExtra, long filesWithLarge,
+ long deletedFiles) {
+ this.statistics = ImmutableMap.copyOf(summary);
+ this.config = config;
+ this.fileStats = new FileStatistics(totalFiles, filesMissingSummary, filesWithExtra, filesWithLarge, deletedFiles);
+ }
+
+ /**
+ * @return Statistics about the files from which summary statistics were obtained.
+ */
+ public FileStatistics getFileStatistics() {
+ return fileStats;
+ }
+
+ /**
+ * @return The configuration used to generate and combine the summary statistics
+ */
+ public SummarizerConfiguration getSummarizerConfiguration() {
+ return config;
+ }
+
+ /**
+ * @return An immutable map of the statistics that were generated and merged by the specified {@link Summarizer}.
+ */
+ public Map<String,Long> getStatistics() {
+ return statistics;
+ }
+
+ @Override
+ public String toString() {
+ return "config : " + config + " filestats : " + fileStats + " statistics : " + statistics;
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/94cdcc4d/core/src/main/java/org/apache/accumulo/core/client/summary/summarizers/DeletesSummarizer.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/summary/summarizers/DeletesSummarizer.java b/core/src/main/java/org/apache/accumulo/core/client/summary/summarizers/DeletesSummarizer.java
new file mode 100644
index 0000000..1e94298
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/client/summary/summarizers/DeletesSummarizer.java
@@ -0,0 +1,75 @@
+/*
+ * 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.apache.accumulo.core.client.summary.summarizers;
+
+import org.apache.accumulo.core.client.admin.TableOperations;
+import org.apache.accumulo.core.client.summary.Summarizer;
+import org.apache.accumulo.core.client.summary.SummarizerConfiguration;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+
+/**
+ * This summarizer tracks the total number of delete Keys seen and the total number of keys seen.
+ *
+ * <p>
+ * This summarizer is used by org.apache.accumulo.tserver.compaction.strategies.TooManyDeletesCompactionStrategy to make compaction decisions based on the
+ * number of deletes.
+ *
+ * @since 2.0.0
+ * @see TableOperations#addSummarizers(String, org.apache.accumulo.core.client.summary.SummarizerConfiguration...)
+ */
+public class DeletesSummarizer implements Summarizer {
+
+ /**
+ * The name of the statistics for the number of deletes.
+ */
+ public static final String DELETES_STAT = "deletes";
+
+ /**
+ * The name of the statistics for the total number of keys.
+ */
+ public static final String TOTAL_STAT = "total";
+
+ @Override
+ public Collector collector(SummarizerConfiguration sc) {
+ return new Collector() {
+
+ long total = 0;
+ long deletes = 0;
+
+ @Override
+ public void accept(Key k, Value v) {
+ total++;
+ if (k.isDeleted()) {
+ deletes++;
+ }
+ }
+
+ @Override
+ public void summarize(StatisticConsumer sc) {
+ sc.accept(DELETES_STAT, deletes);
+ sc.accept(TOTAL_STAT, total);
+ }
+ };
+ }
+
+ @Override
+ public Combiner combiner(SummarizerConfiguration sc) {
+ return (m1, m2) -> m2.forEach((k, v) -> m1.merge(k, v, Long::sum));
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/94cdcc4d/core/src/main/java/org/apache/accumulo/core/client/summary/summarizers/FamilySummarizer.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/summary/summarizers/FamilySummarizer.java b/core/src/main/java/org/apache/accumulo/core/client/summary/summarizers/FamilySummarizer.java
new file mode 100644
index 0000000..9452530
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/client/summary/summarizers/FamilySummarizer.java
@@ -0,0 +1,46 @@
+/*
+ * 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.apache.accumulo.core.client.summary.summarizers;
+
+import java.util.function.UnaryOperator;
+
+import org.apache.accumulo.core.client.admin.TableOperations;
+import org.apache.accumulo.core.client.summary.CountingSummarizer;
+import org.apache.accumulo.core.data.ArrayByteSequence;
+import org.apache.accumulo.core.data.ByteSequence;
+
+/**
+ * Counts column column families. Leverages super class to defend against too many. This class is useful for discovering what column families are present when
+ * the expected number of families is small.
+ *
+ * @since 2.0.0
+ *
+ * @see TableOperations#addSummarizers(String, org.apache.accumulo.core.client.summary.SummarizerConfiguration...)
+ * @see TableOperations#summaries(String)
+ */
+public class FamilySummarizer extends CountingSummarizer<ByteSequence> {
+
+ @Override
+ protected UnaryOperator<ByteSequence> copier() {
+ return ArrayByteSequence::new;
+ }
+
+ @Override
+ protected Converter<ByteSequence> converter() {
+ return (k, v, c) -> c.accept(k.getColumnFamilyData());
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/94cdcc4d/core/src/main/java/org/apache/accumulo/core/client/summary/summarizers/VisibilitySummarizer.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/summary/summarizers/VisibilitySummarizer.java b/core/src/main/java/org/apache/accumulo/core/client/summary/summarizers/VisibilitySummarizer.java
new file mode 100644
index 0000000..c8f76d0
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/client/summary/summarizers/VisibilitySummarizer.java
@@ -0,0 +1,47 @@
+/*
+ * 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.apache.accumulo.core.client.summary.summarizers;
+
+import java.util.function.UnaryOperator;
+
+import org.apache.accumulo.core.client.admin.TableOperations;
+import org.apache.accumulo.core.client.summary.CountingSummarizer;
+import org.apache.accumulo.core.data.ArrayByteSequence;
+import org.apache.accumulo.core.data.ByteSequence;
+
+/**
+ * Counts column visibility labels. Leverages super class to defend against too many. This class is useful for discovering what column visibilities are present
+ * when the expected number of visibilities is small.
+ *
+ * @since 2.0.0
+ *
+ * @see TableOperations#addSummarizers(String, org.apache.accumulo.core.client.summary.SummarizerConfiguration...)
+ * @see TableOperations#summaries(String)
+ */
+public class VisibilitySummarizer extends CountingSummarizer<ByteSequence> {
+
+ @Override
+ protected UnaryOperator<ByteSequence> copier() {
+ return ArrayByteSequence::new;
+ }
+
+ @Override
+ protected Converter<ByteSequence> converter() {
+ return (k, v, c) -> c.accept(k.getColumnVisibilityData());
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/94cdcc4d/core/src/main/java/org/apache/accumulo/core/compaction/CompactionSettings.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/compaction/CompactionSettings.java b/core/src/main/java/org/apache/accumulo/core/compaction/CompactionSettings.java
index 1c5369e..8e65f1c 100644
--- a/core/src/main/java/org/apache/accumulo/core/compaction/CompactionSettings.java
+++ b/core/src/main/java/org/apache/accumulo/core/compaction/CompactionSettings.java
@@ -21,6 +21,8 @@ import java.util.Map;
public enum CompactionSettings {
+ SF_NO_SUMMARY(new NullType()),
+ SF_EXTRA_SUMMARY(new NullType()),
SF_NO_SAMPLE(new NullType()),
SF_GT_ESIZE_OPT(new SizeType()),
SF_LT_ESIZE_OPT(new SizeType()),
http://git-wip-us.apache.org/repos/asf/accumulo/blob/94cdcc4d/core/src/main/java/org/apache/accumulo/core/conf/Property.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
index 6ff2bed..7298db1 100644
--- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java
+++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
@@ -244,6 +244,7 @@ public enum Property {
TSERV_CACHE_POLICY("tserver.cache.policy", "LRU", PropertyType.STRING, "Specifies the eviction policy of the file data caches (LRU or TinyLFU)."),
TSERV_DATACACHE_SIZE("tserver.cache.data.size", "128M", PropertyType.MEMORY, "Specifies the size of the cache for file data blocks."),
TSERV_INDEXCACHE_SIZE("tserver.cache.index.size", "512M", PropertyType.MEMORY, "Specifies the size of the cache for file indices."),
+ TSERV_SUMMARYCACHE_SIZE("tserver.cache.summary.size", "128M", PropertyType.MEMORY, "Specifies the size of the cache for summary data on each tablet server."),
TSERV_PORTSEARCH("tserver.port.search", "false", PropertyType.BOOLEAN, "if the ports above are in use, search higher ports until one is available"),
TSERV_CLIENTPORT("tserver.port.client", "9997", PropertyType.PORT, "The port used for handling client connections on the tablet servers"),
@Deprecated
@@ -359,6 +360,14 @@ public enum Property {
"The number of threads available to load tablets. Recoveries are still performed serially."),
TSERV_SLOW_FLUSH_MILLIS("tserver.slow.flush.time", "100ms", PropertyType.TIMEDURATION,
"If a flush to the write-ahead log takes longer than this period of time, debugging information will written, and may result in a log rollover."),
+ TSERV_SUMMARY_PARTITION_THREADS("tserver.summary.partition.threads", "10", PropertyType.COUNT,
+ "Summary data must be retrieved from files. For a large number of files, the files are broken into partitions of 100K files. This setting determines "
+ + "how many of these groups of 100K files will be processed concurrently."),
+ TSERV_SUMMARY_REMOTE_THREADS("tserver.summary.remote.threads", "128", PropertyType.COUNT,
+ "For a partitioned group of 100K files, those files are grouped by tablet server. Then a remote tablet server is asked to gather summary data. This "
+ + "setting determines how many concurrent request are made per partition."),
+ TSERV_SUMMARY_RETRIEVAL_THREADS("tserver.summary.retrieval.threads", "10", PropertyType.COUNT,
+ "The number of threads on each tablet server available to retrieve summary data, that is not currently in cache, from RFiles."),
// accumulo garbage collector properties
GC_PREFIX("gc.", null, PropertyType.PREFIX, "Properties in this category affect the behavior of the accumulo garbage collector."),
@@ -461,6 +470,9 @@ public enum Property {
"Determines the max # of files each tablet in a table can have. When adjusting this property you may want to consider adjusting"
+ " table.compaction.major.ratio also. Setting this property to 0 will make it default to tserver.scan.files.open.max-1, this will prevent a"
+ " tablet from having more files than can be opened. Setting this property low may throttle ingest and increase query performance."),
+ TABLE_FILE_SUMMARY_MAX_SIZE("table.file.summary.maxSize", "256K", PropertyType.MEMORY, "The maximum size summary that will be stored. The number of"
+ + " files that had summary data exceeding this threshold is reported by Summary.getFileStatistics().getLarge(). When adjusting this"
+ + " consider the expected number files with summaries on each tablet server and the summary cache size."),
@Deprecated
TABLE_WALOG_ENABLED("table.walog.enabled", "true", PropertyType.BOOLEAN, "This setting is deprecated. Use table.durability=none instead."),
TABLE_BLOOM_ENABLED("table.bloom.enabled", "false", PropertyType.BOOLEAN, "Use bloom filters on this table."),
@@ -547,6 +559,13 @@ public enum Property {
TABLE_SUSPEND_DURATION("table.suspend.duration", "0s", PropertyType.TIMEDURATION,
"For tablets belonging to this table: When a tablet server dies, allow the tablet server this duration to revive before reassigning its tablets"
+ "to other tablet servers."),
+ TABLE_SUMMARIZER_PREFIX(
+ "table.summarizer.",
+ null,
+ PropertyType.PREFIX,
+ "Prefix for configuring summarizers for a table. Using this prefix multiple summarizers can be configured with options for each one. Each summarizer configured "
+ + "should have a unique id, this id can be anything. To add a summarizer set table.summarizer.<unique id>=<summarizer class name>. If the summarizer has options, "
+ + "then for each option set table.summarizer.<unique id>.opt.<key>=<value>."),
// VFS ClassLoader properties
VFS_CLASSLOADER_SYSTEM_CLASSPATH_PROPERTY(AccumuloVFSClassLoader.VFS_CLASSLOADER_SYSTEM_CLASSPATH_PROPERTY, "", PropertyType.STRING,
@@ -814,7 +833,8 @@ public enum Property {
return validTableProperties.contains(key) || key.startsWith(Property.TABLE_CONSTRAINT_PREFIX.getKey())
|| key.startsWith(Property.TABLE_ITERATOR_PREFIX.getKey()) || key.startsWith(Property.TABLE_LOCALITY_GROUP_PREFIX.getKey())
|| key.startsWith(Property.TABLE_COMPACTION_STRATEGY_PREFIX.getKey()) || key.startsWith(Property.TABLE_REPLICATION_TARGET.getKey())
- || key.startsWith(Property.TABLE_ARBITRARY_PROP_PREFIX.getKey()) || key.startsWith(TABLE_SAMPLER_OPTS.getKey());
+ || key.startsWith(Property.TABLE_ARBITRARY_PROP_PREFIX.getKey()) || key.startsWith(TABLE_SAMPLER_OPTS.getKey())
+ || key.startsWith(TABLE_SUMMARIZER_PREFIX.getKey());
}
/**
http://git-wip-us.apache.org/repos/asf/accumulo/blob/94cdcc4d/core/src/main/java/org/apache/accumulo/core/conf/PropertyType.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/conf/PropertyType.java b/core/src/main/java/org/apache/accumulo/core/conf/PropertyType.java
index 2c458f0..9726090 100644
--- a/core/src/main/java/org/apache/accumulo/core/conf/PropertyType.java
+++ b/core/src/main/java/org/apache/accumulo/core/conf/PropertyType.java
@@ -19,6 +19,7 @@ package org.apache.accumulo.core.conf;
import static java.util.Objects.requireNonNull;
import java.util.Arrays;
+import java.util.Objects;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.regex.Matcher;
@@ -29,6 +30,8 @@ import org.apache.accumulo.core.util.Pair;
import org.apache.commons.lang.math.IntRange;
import org.apache.hadoop.fs.Path;
+import com.google.common.base.Preconditions;
+
/**
* Types of {@link Property} values. Each type has a short name, a description, and a regex which valid values match. All of these fields are optional.
*/
@@ -89,11 +92,12 @@ public enum PropertyType {
URI("uri", x -> true, "A valid URI");
private String shortname, format;
- private Predicate<String> predicate;
+ // made this transient because findbugs was complaining
+ private transient Predicate<String> predicate;
private PropertyType(String shortname, Predicate<String> predicate, String formatDescription) {
this.shortname = shortname;
- this.predicate = predicate;
+ this.predicate = Objects.requireNonNull(predicate);
this.format = formatDescription;
}
@@ -117,6 +121,7 @@ public enum PropertyType {
* @return true if value is valid or null, or if this type has no regex
*/
public boolean isValidFormat(String value) {
+ Preconditions.checkState(predicate != null, "Predicate was null, maybe this enum was serialized????");
return predicate.test(value);
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/94cdcc4d/core/src/main/java/org/apache/accumulo/core/data/ArrayByteSequence.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/data/ArrayByteSequence.java b/core/src/main/java/org/apache/accumulo/core/data/ArrayByteSequence.java
index 5d16541..bf0ae28 100644
--- a/core/src/main/java/org/apache/accumulo/core/data/ArrayByteSequence.java
+++ b/core/src/main/java/org/apache/accumulo/core/data/ArrayByteSequence.java
@@ -20,6 +20,7 @@ import static java.nio.charset.StandardCharsets.UTF_8;
import java.io.Serializable;
import java.nio.ByteBuffer;
+import java.util.Arrays;
import org.apache.accumulo.core.util.ByteBufferUtil;
@@ -101,6 +102,23 @@ public class ArrayByteSequence extends ByteSequence implements Serializable {
}
}
+ private static byte[] copy(ByteSequence bs) {
+ if (bs.isBackedByArray()) {
+ return Arrays.copyOfRange(bs.getBackingArray(), bs.offset(), bs.offset() + bs.length());
+ } else {
+ return bs.toArray();
+ }
+ }
+
+ /**
+ * Copy constructor. Copies contents of byteSequence.
+ *
+ * @since 2.0.0
+ */
+ public ArrayByteSequence(ByteSequence byteSequence) {
+ this(copy(byteSequence));
+ }
+
@Override
public byte byteAt(int i) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/94cdcc4d/core/src/main/java/org/apache/accumulo/core/data/thrift/TRowRange.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/data/thrift/TRowRange.java b/core/src/main/java/org/apache/accumulo/core/data/thrift/TRowRange.java
new file mode 100644
index 0000000..5d1c062
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/data/thrift/TRowRange.java
@@ -0,0 +1,521 @@
+/*
+ * 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.
+ */
+/**
+ * Autogenerated by Thrift Compiler (0.10.0)
+ *
+ * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
+ * @generated
+ */
+package org.apache.accumulo.core.data.thrift;
+
+@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.10.0)")
+public class TRowRange implements org.apache.thrift.TBase<TRowRange, TRowRange._Fields>, java.io.Serializable, Cloneable, Comparable<TRowRange> {
+ private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("TRowRange");
+
+ private static final org.apache.thrift.protocol.TField START_ROW_FIELD_DESC = new org.apache.thrift.protocol.TField("startRow", org.apache.thrift.protocol.TType.STRING, (short)1);
+ private static final org.apache.thrift.protocol.TField END_ROW_FIELD_DESC = new org.apache.thrift.protocol.TField("endRow", org.apache.thrift.protocol.TType.STRING, (short)2);
+
+ private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new TRowRangeStandardSchemeFactory();
+ private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new TRowRangeTupleSchemeFactory();
+
+ public java.nio.ByteBuffer startRow; // required
+ public java.nio.ByteBuffer endRow; // required
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements org.apache.thrift.TFieldIdEnum {
+ START_ROW((short)1, "startRow"),
+ END_ROW((short)2, "endRow");
+
+ private static final java.util.Map<java.lang.String, _Fields> byName = new java.util.HashMap<java.lang.String, _Fields>();
+
+ static {
+ for (_Fields field : java.util.EnumSet.allOf(_Fields.class)) {
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ public static _Fields findByThriftId(int fieldId) {
+ switch(fieldId) {
+ case 1: // START_ROW
+ return START_ROW;
+ case 2: // END_ROW
+ return END_ROW;
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, throwing an exception
+ * if it is not found.
+ */
+ public static _Fields findByThriftIdOrThrow(int fieldId) {
+ _Fields fields = findByThriftId(fieldId);
+ if (fields == null) throw new java.lang.IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+ return fields;
+ }
+
+ /**
+ * Find the _Fields constant that matches name, or null if its not found.
+ */
+ public static _Fields findByName(java.lang.String name) {
+ return byName.get(name);
+ }
+
+ private final short _thriftId;
+ private final java.lang.String _fieldName;
+
+ _Fields(short thriftId, java.lang.String fieldName) {
+ _thriftId = thriftId;
+ _fieldName = fieldName;
+ }
+
+ public short getThriftFieldId() {
+ return _thriftId;
+ }
+
+ public java.lang.String getFieldName() {
+ return _fieldName;
+ }
+ }
+
+ // isset id assignments
+ public static final java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+ static {
+ java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new java.util.EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
+ tmpMap.put(_Fields.START_ROW, new org.apache.thrift.meta_data.FieldMetaData("startRow", org.apache.thrift.TFieldRequirementType.DEFAULT,
+ new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING , true)));
+ tmpMap.put(_Fields.END_ROW, new org.apache.thrift.meta_data.FieldMetaData("endRow", org.apache.thrift.TFieldRequirementType.DEFAULT,
+ new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING , true)));
+ metaDataMap = java.util.Collections.unmodifiableMap(tmpMap);
+ org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(TRowRange.class, metaDataMap);
+ }
+
+ public TRowRange() {
+ }
+
+ public TRowRange(
+ java.nio.ByteBuffer startRow,
+ java.nio.ByteBuffer endRow)
+ {
+ this();
+ this.startRow = org.apache.thrift.TBaseHelper.copyBinary(startRow);
+ this.endRow = org.apache.thrift.TBaseHelper.copyBinary(endRow);
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public TRowRange(TRowRange other) {
+ if (other.isSetStartRow()) {
+ this.startRow = org.apache.thrift.TBaseHelper.copyBinary(other.startRow);
+ }
+ if (other.isSetEndRow()) {
+ this.endRow = org.apache.thrift.TBaseHelper.copyBinary(other.endRow);
+ }
+ }
+
+ public TRowRange deepCopy() {
+ return new TRowRange(this);
+ }
+
+ @Override
+ public void clear() {
+ this.startRow = null;
+ this.endRow = null;
+ }
+
+ public byte[] getStartRow() {
+ setStartRow(org.apache.thrift.TBaseHelper.rightSize(startRow));
+ return startRow == null ? null : startRow.array();
+ }
+
+ public java.nio.ByteBuffer bufferForStartRow() {
+ return org.apache.thrift.TBaseHelper.copyBinary(startRow);
+ }
+
+ public TRowRange setStartRow(byte[] startRow) {
+ this.startRow = startRow == null ? (java.nio.ByteBuffer)null : java.nio.ByteBuffer.wrap(startRow.clone());
+ return this;
+ }
+
+ public TRowRange setStartRow(java.nio.ByteBuffer startRow) {
+ this.startRow = org.apache.thrift.TBaseHelper.copyBinary(startRow);
+ return this;
+ }
+
+ public void unsetStartRow() {
+ this.startRow = null;
+ }
+
+ /** Returns true if field startRow is set (has been assigned a value) and false otherwise */
+ public boolean isSetStartRow() {
+ return this.startRow != null;
+ }
+
+ public void setStartRowIsSet(boolean value) {
+ if (!value) {
+ this.startRow = null;
+ }
+ }
+
+ public byte[] getEndRow() {
+ setEndRow(org.apache.thrift.TBaseHelper.rightSize(endRow));
+ return endRow == null ? null : endRow.array();
+ }
+
+ public java.nio.ByteBuffer bufferForEndRow() {
+ return org.apache.thrift.TBaseHelper.copyBinary(endRow);
+ }
+
+ public TRowRange setEndRow(byte[] endRow) {
+ this.endRow = endRow == null ? (java.nio.ByteBuffer)null : java.nio.ByteBuffer.wrap(endRow.clone());
+ return this;
+ }
+
+ public TRowRange setEndRow(java.nio.ByteBuffer endRow) {
+ this.endRow = org.apache.thrift.TBaseHelper.copyBinary(endRow);
+ return this;
+ }
+
+ public void unsetEndRow() {
+ this.endRow = null;
+ }
+
+ /** Returns true if field endRow is set (has been assigned a value) and false otherwise */
+ public boolean isSetEndRow() {
+ return this.endRow != null;
+ }
+
+ public void setEndRowIsSet(boolean value) {
+ if (!value) {
+ this.endRow = null;
+ }
+ }
+
+ public void setFieldValue(_Fields field, java.lang.Object value) {
+ switch (field) {
+ case START_ROW:
+ if (value == null) {
+ unsetStartRow();
+ } else {
+ if (value instanceof byte[]) {
+ setStartRow((byte[])value);
+ } else {
+ setStartRow((java.nio.ByteBuffer)value);
+ }
+ }
+ break;
+
+ case END_ROW:
+ if (value == null) {
+ unsetEndRow();
+ } else {
+ if (value instanceof byte[]) {
+ setEndRow((byte[])value);
+ } else {
+ setEndRow((java.nio.ByteBuffer)value);
+ }
+ }
+ break;
+
+ }
+ }
+
+ public java.lang.Object getFieldValue(_Fields field) {
+ switch (field) {
+ case START_ROW:
+ return getStartRow();
+
+ case END_ROW:
+ return getEndRow();
+
+ }
+ throw new java.lang.IllegalStateException();
+ }
+
+ /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+ public boolean isSet(_Fields field) {
+ if (field == null) {
+ throw new java.lang.IllegalArgumentException();
+ }
+
+ switch (field) {
+ case START_ROW:
+ return isSetStartRow();
+ case END_ROW:
+ return isSetEndRow();
+ }
+ throw new java.lang.IllegalStateException();
+ }
+
+ @Override
+ public boolean equals(java.lang.Object that) {
+ if (that == null)
+ return false;
+ if (that instanceof TRowRange)
+ return this.equals((TRowRange)that);
+ return false;
+ }
+
+ public boolean equals(TRowRange that) {
+ if (that == null)
+ return false;
+ if (this == that)
+ return true;
+
+ boolean this_present_startRow = true && this.isSetStartRow();
+ boolean that_present_startRow = true && that.isSetStartRow();
+ if (this_present_startRow || that_present_startRow) {
+ if (!(this_present_startRow && that_present_startRow))
+ return false;
+ if (!this.startRow.equals(that.startRow))
+ return false;
+ }
+
+ boolean this_present_endRow = true && this.isSetEndRow();
+ boolean that_present_endRow = true && that.isSetEndRow();
+ if (this_present_endRow || that_present_endRow) {
+ if (!(this_present_endRow && that_present_endRow))
+ return false;
+ if (!this.endRow.equals(that.endRow))
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int hashCode = 1;
+
+ hashCode = hashCode * 8191 + ((isSetStartRow()) ? 131071 : 524287);
+ if (isSetStartRow())
+ hashCode = hashCode * 8191 + startRow.hashCode();
+
+ hashCode = hashCode * 8191 + ((isSetEndRow()) ? 131071 : 524287);
+ if (isSetEndRow())
+ hashCode = hashCode * 8191 + endRow.hashCode();
+
+ return hashCode;
+ }
+
+ @Override
+ public int compareTo(TRowRange other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+
+ lastComparison = java.lang.Boolean.valueOf(isSetStartRow()).compareTo(other.isSetStartRow());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (isSetStartRow()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.startRow, other.startRow);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ lastComparison = java.lang.Boolean.valueOf(isSetEndRow()).compareTo(other.isSetEndRow());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (isSetEndRow()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.endRow, other.endRow);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ return 0;
+ }
+
+ public _Fields fieldForId(int fieldId) {
+ return _Fields.findByThriftId(fieldId);
+ }
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+ scheme(iprot).read(iprot, this);
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+ scheme(oprot).write(oprot, this);
+ }
+
+ @Override
+ public java.lang.String toString() {
+ java.lang.StringBuilder sb = new java.lang.StringBuilder("TRowRange(");
+ boolean first = true;
+
+ sb.append("startRow:");
+ if (this.startRow == null) {
+ sb.append("null");
+ } else {
+ org.apache.thrift.TBaseHelper.toString(this.startRow, sb);
+ }
+ first = false;
+ if (!first) sb.append(", ");
+ sb.append("endRow:");
+ if (this.endRow == null) {
+ sb.append("null");
+ } else {
+ org.apache.thrift.TBaseHelper.toString(this.endRow, sb);
+ }
+ first = false;
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws org.apache.thrift.TException {
+ // check for required fields
+ // check for sub-struct validity
+ }
+
+ private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+ try {
+ write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, java.lang.ClassNotFoundException {
+ try {
+ read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private static class TRowRangeStandardSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+ public TRowRangeStandardScheme getScheme() {
+ return new TRowRangeStandardScheme();
+ }
+ }
+
+ private static class TRowRangeStandardScheme extends org.apache.thrift.scheme.StandardScheme<TRowRange> {
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot, TRowRange struct) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TField schemeField;
+ iprot.readStructBegin();
+ while (true)
+ {
+ schemeField = iprot.readFieldBegin();
+ if (schemeField.type == org.apache.thrift.protocol.TType.STOP) {
+ break;
+ }
+ switch (schemeField.id) {
+ case 1: // START_ROW
+ if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
+ struct.startRow = iprot.readBinary();
+ struct.setStartRowIsSet(true);
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ break;
+ case 2: // END_ROW
+ if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
+ struct.endRow = iprot.readBinary();
+ struct.setEndRowIsSet(true);
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ break;
+ default:
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ iprot.readFieldEnd();
+ }
+ iprot.readStructEnd();
+
+ // check for required fields of primitive type, which can't be checked in the validate method
+ struct.validate();
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot, TRowRange struct) throws org.apache.thrift.TException {
+ struct.validate();
+
+ oprot.writeStructBegin(STRUCT_DESC);
+ if (struct.startRow != null) {
+ oprot.writeFieldBegin(START_ROW_FIELD_DESC);
+ oprot.writeBinary(struct.startRow);
+ oprot.writeFieldEnd();
+ }
+ if (struct.endRow != null) {
+ oprot.writeFieldBegin(END_ROW_FIELD_DESC);
+ oprot.writeBinary(struct.endRow);
+ oprot.writeFieldEnd();
+ }
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ }
+
+ private static class TRowRangeTupleSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+ public TRowRangeTupleScheme getScheme() {
+ return new TRowRangeTupleScheme();
+ }
+ }
+
+ private static class TRowRangeTupleScheme extends org.apache.thrift.scheme.TupleScheme<TRowRange> {
+
+ @Override
+ public void write(org.apache.thrift.protocol.TProtocol prot, TRowRange struct) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TTupleProtocol oprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+ java.util.BitSet optionals = new java.util.BitSet();
+ if (struct.isSetStartRow()) {
+ optionals.set(0);
+ }
+ if (struct.isSetEndRow()) {
+ optionals.set(1);
+ }
+ oprot.writeBitSet(optionals, 2);
+ if (struct.isSetStartRow()) {
+ oprot.writeBinary(struct.startRow);
+ }
+ if (struct.isSetEndRow()) {
+ oprot.writeBinary(struct.endRow);
+ }
+ }
+
+ @Override
+ public void read(org.apache.thrift.protocol.TProtocol prot, TRowRange struct) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+ java.util.BitSet incoming = iprot.readBitSet(2);
+ if (incoming.get(0)) {
+ struct.startRow = iprot.readBinary();
+ struct.setStartRowIsSet(true);
+ }
+ if (incoming.get(1)) {
+ struct.endRow = iprot.readBinary();
+ struct.setEndRowIsSet(true);
+ }
+ }
+ }
+
+ private static <S extends org.apache.thrift.scheme.IScheme> S scheme(org.apache.thrift.protocol.TProtocol proto) {
+ return (org.apache.thrift.scheme.StandardScheme.class.equals(proto.getScheme()) ? STANDARD_SCHEME_FACTORY : TUPLE_SCHEME_FACTORY).getScheme();
+ }
+ private static void unusedMethod() {}
+}
+