[GitHub] [accumulo] dlmarion commented on a diff in pull request #3232: Proof of Concept for adding conditional mutation support to Ample

["dlmarion (via GitHub)" <gi...@apache.org>]

dlmarion commented on code in PR #3232:
URL: https://github.com/apache/accumulo/pull/3232#discussion_r1140398408


##########
core/src/main/java/org/apache/accumulo/core/metadata/TabletOperation.java:
##########
@@ -0,0 +1,24 @@
+/*
+ * 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;
+
+// TODO where should this go, maybe in MetadataSchema class?
+public enum TabletOperation {
+  NONE, SPLITTING, MERGING, DELETING

Review Comment:
   Do we need STATE_CHANGE for going in between online, offline, ondemand?



##########
core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java:
##########
@@ -303,6 +328,28 @@ TabletMutator putExternalCompaction(ExternalCompactionId ecid,
     void mutate();
   }
 
+  interface ConditionalTabletMutator extends TabletUpdates<ConditionalTabletMutator> {
+
+    ConditionalTabletMutator requireAbsentLocation();
+
+    ConditionalTabletMutator requireLocation(TServerInstance tserver, LocationType type);
+
+    ConditionalTabletMutator requireFile(StoredTabletFile path);
+
+    ConditionalTabletMutator requireAbsentBulikFile(TabletFile bulkref);

Review Comment:
   ```suggestion
       ConditionalTabletMutator requireAbsentBulkFile(TabletFile bulkref);
   ```



##########
server/base/src/main/java/org/apache/accumulo/server/metadata/ConditionalTabletsMutatorImpl.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.server.metadata;
+
+import java.util.*;
+
+import org.apache.accumulo.core.client.ConditionalWriter;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.ConditionalMutation;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.metadata.schema.Ample;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.hadoop.io.Text;
+
+import com.google.common.base.Preconditions;
+
+public class ConditionalTabletsMutatorImpl implements Ample.ConditionalTabletsMutator {
+
+  private final ServerContext context;
+  private TableId currentTableId = null;
+
+  private List<ConditionalMutation> mutations = new ArrayList<>();
+
+  private Map<Text,KeyExtent> extents = new HashMap<>();
+
+  private boolean active = true;
+
+  public ConditionalTabletsMutatorImpl(ServerContext context) {
+    this.context = context;
+  }
+
+  @Override
+  public Ample.ConditionalTabletMutator mutateTablet(KeyExtent extent) {
+    Preconditions.checkState(active);
+    if (currentTableId == null) {
+      currentTableId = extent.tableId();
+    } else if (!currentTableId.equals(extent.tableId())) {
+      throw new IllegalArgumentException(
+          "Can not mix tables ids " + currentTableId + " " + extent.tableId());
+    }
+
+    Preconditions.checkState(extents.putIfAbsent(extent.toMetaRow(), extent) == null,
+        "Duplicate extents not handled");
+    return new ConditionalTabletMutatorImpl(this, context, extent, mutations::add);
+  }
+
+  private ConditionalWriter createConditionalWriter(Ample.DataLevel dataLevel)
+      throws TableNotFoundException {
+    if (dataLevel == Ample.DataLevel.ROOT) {
+      return new RootConditionalWriter(context);
+    } else {
+      return context.createConditionalWriter(dataLevel.metaTable());
+    }
+  }
+
+  @Override
+  public Map<KeyExtent,ConditionalWriter.Result> process() {
+    Preconditions.checkState(active);
+    if (currentTableId != null) {
+      var dataLevel = Ample.DataLevel.of(currentTableId);
+      try (ConditionalWriter conditionalWriter = createConditionalWriter(dataLevel)) {
+        var results = conditionalWriter.write(mutations.iterator());
+
+        var resultsMap = new HashMap<KeyExtent,ConditionalWriter.Result>();
+
+        while (results.hasNext()) {
+          var result = results.next();
+          var row = new Text(result.getMutation().getRow());
+          resultsMap.put(extents.get(row), result);
+        }
+
+        // TODO maybe this check is expensive
+        if (!resultsMap.keySet().equals(Set.copyOf(extents.values()))) {
+          throw new AssertionError("Not all extents were seen, this is unexpected");
+        }
+
+        return resultsMap;
+      } catch (TableNotFoundException e) {
+        throw new RuntimeException(e);
+      } finally {
+        // render inoperable because reuse is not tested
+        extents.clear();
+        mutations.clear();
+        active = false;
+      }
+    } else {
+      // render inoperable because reuse is not tested

Review Comment:
   Should we throw an IllegalStateException instead?



##########
server/base/src/main/java/org/apache/accumulo/server/metadata/RootConditionalWriter.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * 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.server.metadata;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.ConditionalWriter;
+import org.apache.accumulo.core.clientImpl.CompressedIterators;
+import org.apache.accumulo.core.clientImpl.ConditionalWriterImpl;
+import org.apache.accumulo.core.data.ConditionalMutation;
+import org.apache.accumulo.core.dataImpl.thrift.TCMResult;
+import org.apache.accumulo.core.dataImpl.thrift.TConditionalMutation;
+import org.apache.accumulo.core.iteratorsImpl.system.SortedMapIterator;
+import org.apache.accumulo.core.metadata.RootTable;
+import org.apache.accumulo.core.metadata.schema.RootTabletMetadata;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.constraints.MetadataConstraints;
+import org.apache.accumulo.server.data.ServerConditionalMutation;
+import org.apache.accumulo.server.tablets.ConditionCheckerContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A specialized conditional writer that writes metadata for the Accumulo root tablet which is
+ * stored in zookeeper.
+ */
+public class RootConditionalWriter implements ConditionalWriter {
+
+  private static final Logger log = LoggerFactory.getLogger(RootConditionalWriter.class);
+
+  private final ServerContext context;
+
+  RootConditionalWriter(ServerContext context) {
+    this.context = context;
+  }
+
+  @Override
+  public Iterator<Result> write(Iterator<ConditionalMutation> mutations) {
+    List<Result> results = new ArrayList<>();
+    while (mutations.hasNext()) {
+      results.add(write(mutations.next()));

Review Comment:
   I wonder if we should check all mutations for constraint violations before writing, so that we don't end up with a partial write in the event that we have a failure in the middle.



##########
core/src/main/java/org/apache/accumulo/core/metadata/schema/MetadataSchema.java:
##########
@@ -226,6 +226,12 @@ public static void validateDirCol(String dirName) {
        */
       public static final String LOCK_QUAL = "lock";
       public static final ColumnFQ LOCK_COLUMN = new ColumnFQ(NAME, new Text(LOCK_QUAL));
+
+      public static final String OP_QUAL = "op";
+      public static final ColumnFQ OP_COLUMN = new ColumnFQ(NAME, new Text(OP_QUAL));
+
+      public static final String OPID_QUAL = "opid";
+      public static final ColumnFQ OPID_COLUMN = new ColumnFQ(NAME, new Text(OPID_QUAL));

Review Comment:
   Should these go in TabletColumnFamily?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org