Add Mergeability column to support automatic merges

core/src/main/java/org/apache/accumulo/core/client/admin/TabletMergeability.java

@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://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.admin;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.Objects;
+
+import com.google.common.base.Preconditions;
+
+public class TabletMergeability implements Serializable {
+  private static final long serialVersionUID = 1L;
+
+  public static final TabletMergeability NEVER = new TabletMergeability(Duration.ofNanos(-1));
+  public static final TabletMergeability NOW = new TabletMergeability(Duration.ZERO);
+
+  private final Duration delay;
+
+  private TabletMergeability(Duration delay) {
+    this.delay = Objects.requireNonNull(delay);
+  }
+
+  public boolean isNever() {
+    return this.delay.isNegative();
+  }
+
+  public boolean isNow() {
+    return this.delay.isZero();
+  }
+
+  public boolean isFuture() {
+    return delay.toNanos() > 0;
+  }
+
+  public Duration getDelay() {
+    return delay;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    TabletMergeability that = (TabletMergeability) o;
+    return Objects.equals(delay, that.delay);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(delay);
+  }
+
+  @Override
+  public String toString() {
+    if (isNow()) {
+      return "TabletMergeability=NOW";
+    } else if (isNever()) {
+      return "TabletMergeability=NEVER";
+    }
+    return "TabletMergeability=AFTER:" + delay.toNanos();
+  }
+
+  public static TabletMergeability from(Duration delay) {
+    Preconditions.checkArgument(delay.toNanos() >= -1,
+        "Duration of delay must be -1, 0, or a positive offset.");
+    return new TabletMergeability(delay);
+  }
+
+  public static TabletMergeability after(Duration delay) {
+    Preconditions.checkArgument(delay.toNanos() > 0, "Duration of delay must be greater than 0.");
+    return new TabletMergeability(delay);
+  }
+
+}

core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java

@@ -392,6 +392,8 @@ interface TabletUpdates<T> {
 
     T putCloned();
 
+    T putTabletMergeability(TabletMergeabilityMetadata tabletMergeability);
+
     /**
      * By default the server lock is automatically added to mutations unless this method is set to
      * false.

core/src/main/java/org/apache/accumulo/core/metadata/schema/MetadataSchema.java

@@ -158,6 +158,10 @@ public static class TabletColumnFamily {
       public static final String REQUESTED_QUAL = "requestToHost";
       public static final ColumnFQ REQUESTED_COLUMN = new ColumnFQ(NAME, new Text(REQUESTED_QUAL));
 
+      public static final String MERGEABILITY_QUAL = "mergeability";
+      public static final ColumnFQ MERGEABILITY_COLUMN =
+          new ColumnFQ(NAME, new Text(MERGEABILITY_QUAL));
+
       public static Value encodePrevEndRow(Text per) {
         if (per == null) {
           return new Value(new byte[] {0});

core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMergeabilityMetadata.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
+ *
+ *   https://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.metadata.schema;
+
+import static org.apache.accumulo.core.util.LazySingletons.GSON;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.accumulo.core.client.admin.TabletMergeability;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.util.time.SteadyTime;
+
+import com.google.common.base.Preconditions;
+
+public class TabletMergeabilityMetadata implements Serializable {
+  private static final long serialVersionUID = 1L;
+
+  public static final TabletMergeabilityMetadata NEVER =
+      new TabletMergeabilityMetadata(TabletMergeability.NEVER);
+  public static final TabletMergeabilityMetadata NOW =
+      new TabletMergeabilityMetadata(TabletMergeability.NOW);
+
+  private final TabletMergeability tabletMergeability;
+  private final SteadyTime steadyTime;
+
+  private TabletMergeabilityMetadata(TabletMergeability tabletMergeability, SteadyTime steadyTime) {
+    this.tabletMergeability = Objects.requireNonNull(tabletMergeability);
+    this.steadyTime = steadyTime;
+    Preconditions.checkArgument(tabletMergeability.isFuture() == (steadyTime != null),
+        "SteadyTime must be set if and only if TabletMergeability delay is greater than 0.");
+  }
+
+  private TabletMergeabilityMetadata(TabletMergeability tabletMergeability) {
+    this(tabletMergeability, null);
+  }
+
+  public TabletMergeability getTabletMergeability() {
+    return tabletMergeability;
+  }
+
+  public Optional<SteadyTime> getSteadyTime() {
+    return Optional.ofNullable(steadyTime);
+  }
+
+  public boolean isMergeable(SteadyTime currentTime) {
+    if (tabletMergeability.isNever()) {
+      return false;
+    }
+    return currentTime.getDuration().compareTo(totalDelay()) >= 0;
+  }
+
+  private Duration totalDelay() {
+    return steadyTime != null ? steadyTime.getDuration().plus(tabletMergeability.getDelay())
+        : tabletMergeability.getDelay();
+  }
+
+  private static class GSonData {
+    long delay;
+    Long steadyTime;
+  }
+
+  String toJson() {
+    GSonData jData = new GSonData();
+    jData.delay = tabletMergeability.getDelay().toNanos();
+    jData.steadyTime = steadyTime != null ? steadyTime.getNanos() : null;
+    return GSON.get().toJson(jData);
+  }
+
+  static TabletMergeabilityMetadata fromJson(String json) {
+    GSonData jData = GSON.get().fromJson(json, GSonData.class);
+    TabletMergeability tabletMergeability = TabletMergeability.from(Duration.ofNanos(jData.delay));
+    SteadyTime steadyTime =
+        jData.steadyTime != null ? SteadyTime.from(jData.steadyTime, TimeUnit.NANOSECONDS) : null;
+    return new TabletMergeabilityMetadata(tabletMergeability, steadyTime);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    TabletMergeabilityMetadata that = (TabletMergeabilityMetadata) o;
+    return Objects.equals(tabletMergeability, that.tabletMergeability)
+        && Objects.equals(steadyTime, that.steadyTime);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(tabletMergeability, steadyTime);
+  }
+
+  @Override
+  public String toString() {
+    return "TabletMergeabilityMetadata{" + tabletMergeability + ", " + steadyTime + '}';
+  }
+
+  public static TabletMergeabilityMetadata after(Duration delay, SteadyTime currentTime) {
+    return from(TabletMergeability.after(delay), currentTime);
+  }
+
+  public static TabletMergeabilityMetadata from(TabletMergeability tm, SteadyTime currentTime) {
+    return new TabletMergeabilityMetadata(tm, currentTime);
+  }
+
+  public static Value toValue(TabletMergeabilityMetadata tmm) {
+    return new Value(tmm.toJson());
+  }
+
+  public static TabletMergeabilityMetadata fromValue(Value value) {
+    return TabletMergeabilityMetadata.fromJson(value.toString());
+  }
+}

core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java

@@ -25,6 +25,7 @@
 import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.SELECTED_QUAL;
 import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.TIME_QUAL;
 import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily.AVAILABILITY_QUAL;
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily.MERGEABILITY_QUAL;
 import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily.PREV_ROW_QUAL;
 import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily.REQUESTED_QUAL;
 
@@ -123,6 +124,7 @@ public class TabletMetadata {
   private final Set<FateId> compacted;
   private final Set<FateId> userCompactionsRequested;
   private final UnSplittableMetadata unSplittableMetadata;
+  private final TabletMergeabilityMetadata mergeability;
   private final Supplier<Long> fileSize;
 
   private TabletMetadata(Builder tmBuilder) {
@@ -155,6 +157,7 @@ private TabletMetadata(Builder tmBuilder) {
     this.compacted = tmBuilder.compacted.build();
     this.userCompactionsRequested = tmBuilder.userCompactionsRequested.build();
     this.unSplittableMetadata = tmBuilder.unSplittableMetadata;
+    this.mergeability = Objects.requireNonNull(tmBuilder.mergeability);
     this.fileSize = Suppliers.memoize(() -> {
       // This code was using a java stream. While profiling SplitMillionIT, the stream was showing
       // up as hot when scanning 1 million tablets. Converted to a for loop to improve performance.
@@ -198,7 +201,8 @@ public enum ColumnType {
     SELECTED,
     COMPACTED,
     USER_COMPACTION_REQUESTED,
-    UNSPLITTABLE
+    UNSPLITTABLE,
+    MERGEABILITY
   }
 
   public static class Location {
@@ -439,6 +443,11 @@ public UnSplittableMetadata getUnSplittable() {
     return unSplittableMetadata;
   }
 
+  public TabletMergeabilityMetadata getTabletMergeability() {
+    ensureFetched(ColumnType.MERGEABILITY);
+    return mergeability;
+  }
+
   @Override
   public String toString() {
     return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE).append("tableId", tableId)
@@ -453,7 +462,8 @@ public String toString() {
         .append("operationId", operationId).append("selectedFiles", selectedFiles)
         .append("futureAndCurrentLocationSet", futureAndCurrentLocationSet)
         .append("userCompactionsRequested", userCompactionsRequested)
-        .append("unSplittableMetadata", unSplittableMetadata).toString();
+        .append("unSplittableMetadata", unSplittableMetadata).append("mergeability", mergeability)
+        .toString();
   }
 
   public List<Entry<Key,Value>> getKeyValues() {
@@ -527,6 +537,9 @@ public static <E extends Entry<Key,Value>> TabletMetadata convertRow(Iterator<E>
             case REQUESTED_QUAL:
               tmBuilder.onDemandHostingRequested(true);
               break;
+            case MERGEABILITY_QUAL:
+              tmBuilder.mergeability(TabletMergeabilityMetadata.fromValue(kv.getValue()));
+              break;
             default:
               throw new IllegalStateException("Unexpected TabletColumnFamily qualifier: " + qual);
           }
@@ -689,7 +702,7 @@ static class Builder {
     private final ImmutableSet.Builder<FateId> compacted = ImmutableSet.builder();
     private final ImmutableSet.Builder<FateId> userCompactionsRequested = ImmutableSet.builder();
     private UnSplittableMetadata unSplittableMetadata;
-    // private Supplier<Long> fileSize;
+    private TabletMergeabilityMetadata mergeability = TabletMergeabilityMetadata.NEVER;
 
     void table(TableId tableId) {
       this.tableId = tableId;
@@ -799,6 +812,10 @@ void unSplittableMetadata(UnSplittableMetadata unSplittableMetadata) {
       this.unSplittableMetadata = unSplittableMetadata;
     }
 
+    void mergeability(TabletMergeabilityMetadata mergeability) {
+      this.mergeability = mergeability;
+    }
+
     void keyValue(Entry<Key,Value> kv) {
       if (this.keyValues == null) {
         this.keyValues = ImmutableList.builder();

core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadataBuilder.java

@@ -30,6 +30,7 @@
 import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.LOADED;
 import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.LOCATION;
 import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.LOGS;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.MERGEABILITY;
 import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.MERGED;
 import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.OPID;
 import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.PREV_ROW;
@@ -312,6 +313,14 @@ public TabletMetadataBuilder putCloned() {
     return this;
   }
 
+  @Override
+  public TabletMetadataBuilder
+      putTabletMergeability(TabletMergeabilityMetadata tabletMergeability) {
+    fetched.add(MERGEABILITY);
+    internalBuilder.putTabletMergeability(tabletMergeability);
+    return this;
+  }
+
   @Override
   public TabletMetadataBuilder automaticallyPutServerLock(boolean b) {
     throw new UnsupportedOperationException();

core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMutatorBase.java

@@ -390,6 +390,13 @@ public T automaticallyPutServerLock(boolean b) {
     return getThis();
   }
 
+  @Override
+  public T putTabletMergeability(TabletMergeabilityMetadata tabletMergeability) {
+    TabletColumnFamily.MERGEABILITY_COLUMN.put(mutation,
+        TabletMergeabilityMetadata.toValue(tabletMergeability));
+    return getThis();
+  }
+
   public void setCloseAfterMutate(AutoCloseable closeable) {
     this.closeAfterMutate = closeable;
   }

core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java

@@ -77,6 +77,7 @@
 import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ScanFileColumnFamily;
 import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.SplitColumnFamily;
 import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.SuspendLocationColumn;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
 import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.UserCompactionRequestedColumnFamily;
 import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType;
 import org.apache.accumulo.core.metadata.schema.filters.TabletMetadataFilter;
@@ -394,6 +395,9 @@ public Options fetch(ColumnType... colsToFetch) {
           case UNSPLITTABLE:
             qualifiers.add(SplitColumnFamily.UNSPLITTABLE_COLUMN);
             break;
+          case MERGEABILITY:
+            qualifiers.add(TabletColumnFamily.MERGEABILITY_COLUMN);
+            break;
           default:
             throw new IllegalArgumentException("Unknown col type " + colToFetch);
         }