[#663] improvement(core): Add storage layout version information and …

…store it in kv store (#701)

### What changes were proposed in this pull request?

Save the KV layout version information into KV storage.

### Why are the changes needed?

We require storage layout version information to determine the code path
we should use.


Fix: #663 

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

New UTs in `TestKvEntityStore`

core/src/main/java/com/datastrato/gravitino/storage/StorageLayoutException.java

@@ -0,0 +1,19 @@
+/*
+ * Copyright 2023 Datastrato.
+ * This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.storage;
+
+import com.datastrato.gravitino.exceptions.GravitinoRuntimeException;
+
+public class StorageLayoutException extends GravitinoRuntimeException {
+
+  public StorageLayoutException(String message) {
+    super(message);
+  }
+
+  public StorageLayoutException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}

core/src/main/java/com/datastrato/gravitino/storage/StorageLayoutVersion.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2023 Datastrato.
+ * This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.storage;
+
+import lombok.Getter;
+
+/** The storage layer version of the entity store. */
+@Getter
+public enum StorageLayoutVersion {
+  V1("v1");
+
+  private final String version;
+
+  StorageLayoutVersion(String version) {
+    this.version = version;
+  }
+
+  public static StorageLayoutVersion fromString(String version) {
+    for (StorageLayoutVersion v : StorageLayoutVersion.values()) {
+      if (v.version.equals(version)) {
+        return v;
+      }
+    }
+    throw new StorageLayoutException(
+        "Unknown storage version, maybe the data is broken, please "
+            + "check the storage directory.");
+  }
+}

core/src/main/java/com/datastrato/gravitino/storage/kv/KvEntityStore.java

@@ -26,6 +26,7 @@
 import com.datastrato.gravitino.exceptions.NonEmptyEntityException;
 import com.datastrato.gravitino.storage.EntityKeyEncoder;
 import com.datastrato.gravitino.storage.NameMappingService;
+import com.datastrato.gravitino.storage.StorageLayoutVersion;
 import com.datastrato.gravitino.utils.Bytes;
 import com.datastrato.gravitino.utils.Executable;
 import com.google.common.annotations.VisibleForTesting;
@@ -58,6 +59,7 @@ public class KvEntityStore implements EntityStore {
   public static final Logger LOGGER = LoggerFactory.getLogger(KvEntityStore.class);
   public static final ImmutableMap<String, String> KV_BACKENDS =
       ImmutableMap.of("RocksDBKvBackend", RocksDBKvBackend.class.getCanonicalName());
+  public static final String LAYOUT_VERSION = "layout_version";
 
   @Getter @VisibleForTesting private KvBackend backend;
 
@@ -67,6 +69,11 @@ public class KvEntityStore implements EntityStore {
   private EntityKeyEncoder<byte[]> entityKeyEncoder;
   private NameMappingService nameMappingService;
   private EntitySerDe serDe;
+  // We will use storageLayoutVersion to check whether the layout of the storage is compatible with
+  // the current version of the code.
+  // Note: If we change the layout of the storage in the future, please update the value of
+  // storageLayoutVersion if it's necessary.
+  @VisibleForTesting StorageLayoutVersion storageLayoutVersion;
 
   @Override
   public void initialize(Config config) throws RuntimeException {
@@ -76,6 +83,7 @@ public void initialize(Config config) throws RuntimeException {
     this.nameMappingService = new KvNameMappingService(backend);
     this.entityKeyEncoder = new BinaryEntityKeyEncoder(nameMappingService);
     this.reentrantReadWriteLock = new ReentrantReadWriteLock();
+    this.storageLayoutVersion = initStorageVersionInfo();
   }
 
   @Override
@@ -383,6 +391,25 @@ private static KvBackend createKvEntityBackend(Config config) {
     }
   }
 
+  private StorageLayoutVersion initStorageVersionInfo() {
+    byte[] bytes;
+    try {
+      bytes = backend.get(LAYOUT_VERSION.getBytes(StandardCharsets.UTF_8));
+      if (bytes == null) {
+        // If the layout version is not set, we will set it to the default version.
+        backend.put(
+            LAYOUT_VERSION.getBytes(StandardCharsets.UTF_8),
+            StorageLayoutVersion.V1.getVersion().getBytes(StandardCharsets.UTF_8),
+            true);
+        return StorageLayoutVersion.V1;
+      }
+
+      return StorageLayoutVersion.fromString(new String(bytes));
+    } catch (IOException e) {
+      throw new IllegalStateException("Failed to get/put layout version information", e);
+    }
+  }
+
   @FunctionalInterface
   interface IOExecutable<R> {
     R execute() throws IOException;

core/src/test/java/com/datastrato/gravitino/storage/TestStorageVersion.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2023 Datastrato.
+ * This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.storage;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestStorageVersion {
+
+  @Test
+  void testFromString() {
+    StorageLayoutVersion version = StorageLayoutVersion.fromString("v1");
+    Assertions.assertEquals(StorageLayoutVersion.V1, version);
+
+    Assertions.assertThrowsExactly(
+        StorageLayoutException.class, () -> StorageLayoutVersion.fromString("v200000.0"));
+  }
+}

core/src/test/java/com/datastrato/gravitino/storage/kv/TestKvEntityStorage.java

@@ -30,6 +30,7 @@
 import com.datastrato.gravitino.meta.SchemaEntity;
 import com.datastrato.gravitino.meta.SchemaVersion;
 import com.datastrato.gravitino.meta.TableEntity;
+import com.datastrato.gravitino.storage.StorageLayoutVersion;
 import com.google.common.io.Files;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.io.File;
@@ -857,4 +858,33 @@ void testConcurrentIssues() throws IOException, ExecutionException, InterruptedE
       Assertions.assertEquals(9, totalFailed);
     }
   }
+
+  @Test
+  void testStorageLayoutVersion() throws IOException {
+    Config config = Mockito.mock(Config.class);
+    File file = Files.createTempDir();
+    file.deleteOnExit();
+    Mockito.when(config.get(ENTITY_STORE)).thenReturn("kv");
+    Mockito.when(config.get(ENTITY_KV_STORE)).thenReturn(DEFAULT_ENTITY_KV_STORE);
+    Mockito.when(config.get(Configs.ENTITY_SERDE)).thenReturn("proto");
+    Mockito.when(config.get(ENTRY_KV_ROCKSDB_BACKEND_PATH)).thenReturn(file.getAbsolutePath());
+
+    // First time create entity store, the storage layout version should be DEFAULT_LAYOUT_VERSION
+    try (EntityStore store = EntityStoreFactory.createEntityStore(config)) {
+      store.initialize(config);
+      Assertions.assertTrue(store instanceof KvEntityStore);
+      store.setSerDe(EntitySerDeFactory.createEntitySerDe(config.get(Configs.ENTITY_SERDE)));
+      KvEntityStore entityStore = (KvEntityStore) store;
+      Assertions.assertEquals(StorageLayoutVersion.V1, entityStore.storageLayoutVersion);
+    }
+
+    // Second time create entity store, the storage layout version should be DEFAULT_LAYOUT_VERSION
+    try (EntityStore store = EntityStoreFactory.createEntityStore(config)) {
+      store.initialize(config);
+      Assertions.assertTrue(store instanceof KvEntityStore);
+      store.setSerDe(EntitySerDeFactory.createEntitySerDe(config.get(Configs.ENTITY_SERDE)));
+      KvEntityStore entityStore = (KvEntityStore) store;
+      Assertions.assertEquals(StorageLayoutVersion.V1, entityStore.storageLayoutVersion);
+    }
+  }
 }