HDDS-748. Use strongly typed metadata Table implementation. Contribut…

…ed by Elek Marton.

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/ByteArrayKeyValue.java

@@ -0,0 +1,67 @@
+/*
+ * 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.hadoop.utils.db;
+
+import org.apache.hadoop.utils.db.Table.KeyValue;
+
+/**
+ * Key value for raw Table implementations.
+ */
+public final class ByteArrayKeyValue implements KeyValue<byte[], byte[]> {
+  private byte[] key;
+  private byte[] value;
+
+  private ByteArrayKeyValue(byte[] key, byte[] value) {
+    this.key = key;
+    this.value = value;
+  }
+
+  /**
+   * Create a KeyValue pair.
+   *
+   * @param key   - Key Bytes
+   * @param value - Value bytes
+   * @return KeyValue object.
+   */
+  public static ByteArrayKeyValue create(byte[] key, byte[] value) {
+    return new ByteArrayKeyValue(key, value);
+  }
+
+  /**
+   * Return key.
+   *
+   * @return byte[]
+   */
+  public byte[] getKey() {
+    byte[] result = new byte[key.length];
+    System.arraycopy(key, 0, result, 0, key.length);
+    return result;
+  }
+
+  /**
+   * Return value.
+   *
+   * @return byte[]
+   */
+  public byte[] getValue() {
+    byte[] result = new byte[value.length];
+    System.arraycopy(value, 0, result, 0, value.length);
+    return result;
+  }
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/Codec.java

@@ -0,0 +1,38 @@
+/*
+ * 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.hadoop.utils.db;
+
+/**
+ * Codec interface to marshall/unmarshall data to/from a byte[] based
+ * key/value store.
+ *
+ * @param <T> Unserialized type
+ */
+public interface Codec<T> {
+
+  /**
+   * Convert object to raw persisted format.
+   */
+  byte[] toPersistedFormat(T object);
+
+  /**
+   * Convert object from raw persisted format.
+   */
+  T fromPersistedFormat(byte[] rawData);
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/CodecRegistry.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.apache.hadoop.utils.db;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Collection of available codecs.
+ */
+public class CodecRegistry {
+
+  private Map<Class, Codec<?>> valueCodecs;
+
+  public CodecRegistry() {
+    valueCodecs = new HashMap<>();
+    valueCodecs.put(String.class, new StringCodec());
+  }
+
+  /**
+   * Convert raw value to strongly typed value/key with the help of a codec.
+   *
+   * @param rawData original byte array from the db.
+   * @param format  Class of the return value
+   * @param <T>     Type of the return value.
+   * @return the object with the parsed field data
+   */
+  public <T> T asObject(byte[] rawData, Class<T> format) {
+    if (valueCodecs.containsKey(format)) {
+      return (T) valueCodecs.get(format).fromPersistedFormat(rawData);
+    } else {
+      throw new IllegalStateException(
+          "Codec is not registered for type: " + format);
+    }
+  }
+
+  /**
+   * Convert strongly typed object to raw data to store it in the kv store.
+   *
+   * @param object typed object.
+   * @param <T>    Type of the typed object.
+   * @return byte array to store it ini the kv store.
+   */
+  public <T> byte[] asRawData(T object) {
+    Class<T> format = (Class<T>) object.getClass();
+    if (valueCodecs.containsKey(format)) {
+      Codec<T> codec = (Codec<T>) valueCodecs.get(format);
+      return codec.toPersistedFormat(object);
+    } else {
+      throw new IllegalStateException(
+          "Codec is not registered for type: " + format);
+    }
+  }
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/DBStore.java

@@ -41,7 +41,17 @@ public interface DBStore extends AutoCloseable {
    * @return - TableStore.
    * @throws IOException on Failure
    */
-  Table getTable(String name) throws IOException;
+  Table<byte[], byte[]> getTable(String name) throws IOException;
+
+  /**
+   * Gets an existing TableStore with implicit key/value conversion.
+   *
+   * @param name - Name of the TableStore to get
+   * @return - TableStore.
+   * @throws IOException on Failure
+   */
+  <KEY, VALUE> Table<KEY, VALUE> getTable(String name,
+      Class<KEY> keyType, Class<VALUE> valueType) throws IOException;
 
   /**
    * Lists the Known list of Tables in a DB.

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/RDBStore.java

@@ -55,6 +55,7 @@ public class RDBStore implements DBStore {
   private final File dbLocation;
   private final WriteOptions writeOptions;
   private final DBOptions dbOptions;
+  private final CodecRegistry codecRegistry;
   private final Hashtable<String, ColumnFamilyHandle> handleTable;
   private ObjectName statMBeanName;
 
@@ -64,7 +65,7 @@ public RDBStore(File dbFile, DBOptions options, Set<TableConfig> families)
     Preconditions.checkNotNull(families);
     Preconditions.checkArgument(families.size() > 0);
     handleTable = new Hashtable<>();
-
+    codecRegistry = new CodecRegistry();
     final List<ColumnFamilyDescriptor> columnFamilyDescriptors =
         new ArrayList<>();
     final List<ColumnFamilyHandle> columnFamilyHandles = new ArrayList<>();
@@ -254,14 +255,21 @@ protected ObjectName getStatMBeanName() {
   }
 
   @Override
-  public Table getTable(String name) throws IOException {
+  public Table<byte[], byte[]> getTable(String name) throws IOException {
     ColumnFamilyHandle handle = handleTable.get(name);
     if (handle == null) {
       throw new IOException("No such table in this DB. TableName : " + name);
     }
     return new RDBTable(this.db, handle, this.writeOptions);
   }
 
+  @Override
+  public <KEY, VALUE> Table<KEY, VALUE> getTable(String name,
+      Class<KEY> keyType, Class<VALUE> valueType) throws IOException {
+    return new TypedTable<KEY, VALUE>(getTable(name), codecRegistry, keyType,
+        valueType);
+  }
+
   @Override
   public ArrayList<Table> listTables() throws IOException {
     ArrayList<Table> returnList = new ArrayList<>();

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/RDBStoreIterator.java

@@ -19,17 +19,17 @@
 
 package org.apache.hadoop.utils.db;
 
-import org.apache.hadoop.utils.db.Table.KeyValue;
-import org.rocksdb.RocksIterator;
-
 import java.io.IOException;
 import java.util.NoSuchElementException;
 import java.util.function.Consumer;
 
+import org.rocksdb.RocksIterator;
+
 /**
  * RocksDB store iterator.
  */
-public class RDBStoreIterator implements TableIterator<KeyValue> {
+public class RDBStoreIterator
+    implements TableIterator<byte[], ByteArrayKeyValue> {
 
   private RocksIterator rocksDBIterator;
 
@@ -39,7 +39,8 @@ public RDBStoreIterator(RocksIterator iterator) {
   }
 
   @Override
-  public void forEachRemaining(Consumer<? super KeyValue> action) {
+  public void forEachRemaining(
+      Consumer<? super ByteArrayKeyValue> action) {
     while(hasNext()) {
       action.accept(next());
     }
@@ -51,9 +52,10 @@ public boolean hasNext() {
   }
 
   @Override
-  public Table.KeyValue next() {
+  public ByteArrayKeyValue next() {
     if (rocksDBIterator.isValid()) {
-      KeyValue value = KeyValue.create(rocksDBIterator.key(), rocksDBIterator
+      ByteArrayKeyValue value =
+          ByteArrayKeyValue.create(rocksDBIterator.key(), rocksDBIterator
           .value());
       rocksDBIterator.next();
       return value;
@@ -72,10 +74,10 @@ public void seekToLast() {
   }
 
   @Override
-  public KeyValue seek(byte[] key) {
+  public ByteArrayKeyValue seek(byte[] key) {
     rocksDBIterator.seek(key);
     if (rocksDBIterator.isValid()) {
-      return KeyValue.create(rocksDBIterator.key(),
+      return ByteArrayKeyValue.create(rocksDBIterator.key(),
           rocksDBIterator.value());
     }
     return null;

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/RDBTable.java

@@ -35,7 +35,8 @@
 /**
  * RocksDB implementation of ozone metadata store.
  */
-public class RDBTable implements Table {
+public class RDBTable implements Table<byte[], byte[]> {
+
 
   private static final Logger LOG =
       LoggerFactory.getLogger(RDBTable.class);
@@ -108,7 +109,7 @@ public void putWithBatch(BatchOperation batch, byte[] key, byte[] value)
 
   @Override
   public boolean isEmpty() throws IOException {
-    try (TableIterator<KeyValue> keyIter = iterator()) {
+    try (TableIterator<byte[], ByteArrayKeyValue> keyIter = iterator()) {
       keyIter.seekToFirst();
       return !keyIter.hasNext();
     }
@@ -145,7 +146,7 @@ public void deleteWithBatch(BatchOperation batch, byte[] key)
   }
 
   @Override
-  public TableIterator<KeyValue> iterator() {
+  public TableIterator<byte[], ByteArrayKeyValue> iterator() {
     ReadOptions readOptions = new ReadOptions();
     return new RDBStoreIterator(db.newIterator(handle, readOptions));
   }

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/StringCodec.java

@@ -0,0 +1,45 @@
+/*
+ * 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.hadoop.utils.db;
+
+import org.apache.hadoop.hdfs.DFSUtil;
+
+/**
+ * Codec to convert String to/from byte array.
+ */
+public class StringCodec implements Codec<String> {
+
+  @Override
+  public byte[] toPersistedFormat(String object) {
+    if (object != null) {
+      return DFSUtil.string2Bytes(object);
+    } else {
+      return null;
+    }
+  }
+
+  @Override
+  public String fromPersistedFormat(byte[] rawData) {
+    if (rawData != null) {
+      return DFSUtil.bytes2String(rawData);
+    } else {
+      return null;
+    }
+  }
+}