// Copyright (c) 2014, Facebook, Inc.  All rights reserved.
// This source code is licensed under the BSD-style license found in the
// LICENSE file in the root directory of this source tree. An additional grant
// of patent rights can be found in the PATENTS file in the same directory.

package org.rocksdb.test;

import java.util.HashMap;
import java.util.List;
import java.util.ArrayList;
import java.util.Map;

import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.rocksdb.*;

import static org.assertj.core.api.Assertions.assertThat;

public class ColumnFamilyTest {

  @ClassRule
  public static final RocksMemoryResource rocksMemoryResource =
      new RocksMemoryResource();

  @Rule
  public TemporaryFolder dbFolder = new TemporaryFolder();

  @Test
  public void listColumnFamilies() throws RocksDBException {
    RocksDB db = null;
    Options options = null;
    try {
      options = new Options();
      options.setCreateIfMissing(true);

      DBOptions dbOptions = new DBOptions();
      dbOptions.setCreateIfMissing(true);

      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath());
      // Test listColumnFamilies
      List<byte[]> columnFamilyNames;
      columnFamilyNames = RocksDB.listColumnFamilies(options, dbFolder.getRoot().getAbsolutePath());
      assertThat(columnFamilyNames).isNotNull();
      assertThat(columnFamilyNames.size()).isGreaterThan(0);
      assertThat(columnFamilyNames.size()).isEqualTo(1);
      assertThat(new String(columnFamilyNames.get(0))).isEqualTo("default");
    } finally {
      if (db != null) {
        db.close();
      }
      if (options != null) {
        options.dispose();
      }
    }
  }

  @Test
  public void createColumnFamily() throws RocksDBException {
    RocksDB db = null;
    Options options = null;
    try {
      options = new Options();
      options.setCreateIfMissing(true);

      DBOptions dbOptions = new DBOptions();
      dbOptions.setCreateIfMissing(true);

      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath());
      db.createColumnFamily(new ColumnFamilyDescriptor("new_cf",
          new ColumnFamilyOptions()));
      db.close();
      List<byte[]> columnFamilyNames;
      columnFamilyNames = RocksDB.listColumnFamilies(options, dbFolder.getRoot().getAbsolutePath());
      assertThat(columnFamilyNames).isNotNull();
      assertThat(columnFamilyNames.size()).isGreaterThan(0);
      assertThat(columnFamilyNames.size()).isEqualTo(2);
      assertThat(new String(columnFamilyNames.get(0))).isEqualTo("default");
      assertThat(new String(columnFamilyNames.get(1))).isEqualTo("new_cf");
    } finally {
      if (db != null) {
        db.close();
      }
      if (options != null) {
        options.dispose();
      }
    }
  }

  @Test
  public void openWithColumnFamilies() throws RocksDBException {
    RocksDB db = null;
    DBOptions options = null;
    try {
      options = new DBOptions();
      options.setCreateIfMissing(true);
      options.setCreateMissingColumnFamilies(true);
      // Test open database with column family names
      List<ColumnFamilyDescriptor> cfNames =
          new ArrayList<>();
      List<ColumnFamilyHandle> columnFamilyHandleList =
          new ArrayList<>();
      cfNames.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
      cfNames.add(new ColumnFamilyDescriptor("new_cf"));

      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(),
          cfNames, columnFamilyHandleList);
      assertThat(columnFamilyHandleList.size()).isEqualTo(2);
      db.put("dfkey1".getBytes(), "dfvalue".getBytes());
      db.put(columnFamilyHandleList.get(0), "dfkey2".getBytes(),
          "dfvalue".getBytes());
      db.put(columnFamilyHandleList.get(1), "newcfkey1".getBytes(),
          "newcfvalue".getBytes());

      String retVal = new String(db.get(columnFamilyHandleList.get(1),
          "newcfkey1".getBytes()));
      assertThat(retVal).isEqualTo("newcfvalue");
      assertThat((db.get(columnFamilyHandleList.get(1),
          "dfkey1".getBytes()))).isNull();
      db.remove(columnFamilyHandleList.get(1), "newcfkey1".getBytes());
      assertThat((db.get(columnFamilyHandleList.get(1),
          "newcfkey1".getBytes()))).isNull();
      db.remove(columnFamilyHandleList.get(0), new WriteOptions(),
          "dfkey2".getBytes());
      assertThat(db.get(columnFamilyHandleList.get(0), new ReadOptions(),
          "dfkey2".getBytes())).isNull();
    } finally {
      if (db != null) {
        db.close();
      }
      if (options != null) {
        options.dispose();
      }
    }
  }

  @Test
  public void getWithOutValueAndCf() throws RocksDBException {
    RocksDB db = null;
    DBOptions options = null;
    try {
      options = new DBOptions();
      options.setCreateIfMissing(true);
      options.setCreateMissingColumnFamilies(true);
      // Test open database with column family names
      List<ColumnFamilyDescriptor> cfDescriptors =
          new ArrayList<>();
      List<ColumnFamilyHandle> columnFamilyHandleList =
          new ArrayList<>();
      cfDescriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(),
          cfDescriptors, columnFamilyHandleList);
      db.put(columnFamilyHandleList.get(0), new WriteOptions(),
          "key1".getBytes(), "value".getBytes());
      db.put("key2".getBytes(), "12345678".getBytes());
      byte[] outValue = new byte[5];
      // not found value
      int getResult = db.get("keyNotFound".getBytes(), outValue);
      assertThat(getResult).isEqualTo(RocksDB.NOT_FOUND);
      // found value which fits in outValue
      getResult = db.get(columnFamilyHandleList.get(0), "key1".getBytes(), outValue);
      assertThat(getResult).isNotEqualTo(RocksDB.NOT_FOUND);
      assertThat(outValue).isEqualTo("value".getBytes());
      // found value which fits partially
      getResult = db.get(columnFamilyHandleList.get(0), new ReadOptions(),
          "key2".getBytes(), outValue);
      assertThat(getResult).isNotEqualTo(RocksDB.NOT_FOUND);
      assertThat(outValue).isEqualTo("12345".getBytes());
    } finally {
      if (db != null) {
        db.close();
      }
      if (options != null) {
        options.dispose();
      }
    }
  }

  @Test
  public void createWriteDropColumnFamily() throws RocksDBException {
    RocksDB db = null;
    DBOptions opt = null;
    ColumnFamilyHandle tmpColumnFamilyHandle = null;
    try {
      opt = new DBOptions();
      opt.setCreateIfMissing(true);
      opt.setCreateMissingColumnFamilies(true);
      List<ColumnFamilyDescriptor> cfNames =
          new ArrayList<>();
      List<ColumnFamilyHandle> columnFamilyHandleList =
          new ArrayList<>();
      cfNames.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
      cfNames.add(new ColumnFamilyDescriptor("new_cf"));

      db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath(),
          cfNames, columnFamilyHandleList);
      tmpColumnFamilyHandle = db.createColumnFamily(
          new ColumnFamilyDescriptor("tmpCF", new ColumnFamilyOptions()));
      db.put(tmpColumnFamilyHandle, "key".getBytes(), "value".getBytes());
      db.dropColumnFamily(tmpColumnFamilyHandle);
      tmpColumnFamilyHandle.dispose();
    } finally {
      if (tmpColumnFamilyHandle != null) {
        tmpColumnFamilyHandle.dispose();
      }
      if (db != null) {
        db.close();
      }
      if (opt != null) {
        opt.dispose();
      }
    }
  }

  @Test
  public void writeBatch() throws RocksDBException {
    RocksDB db = null;
    DBOptions opt = null;
    try {
      opt = new DBOptions();
      opt.setCreateIfMissing(true);
      opt.setCreateMissingColumnFamilies(true);
      List<ColumnFamilyDescriptor> cfNames =
          new ArrayList<>();
      List<ColumnFamilyHandle> columnFamilyHandleList =
          new ArrayList<>();
      cfNames.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
      cfNames.add(new ColumnFamilyDescriptor("new_cf"));

      db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath(),
          cfNames, columnFamilyHandleList);

      WriteBatch writeBatch = new WriteBatch();
      WriteOptions writeOpt = new WriteOptions();
      writeBatch.put("key".getBytes(), "value".getBytes());
      writeBatch.put(columnFamilyHandleList.get(1), "newcfkey".getBytes(),
          "value".getBytes());
      writeBatch.put(columnFamilyHandleList.get(1), "newcfkey2".getBytes(),
          "value2".getBytes());
      writeBatch.remove("xyz".getBytes());
      writeBatch.remove(columnFamilyHandleList.get(1), "xyz".getBytes());
      db.write(writeOpt, writeBatch);
      writeBatch.dispose();
      assertThat(db.get(columnFamilyHandleList.get(1),
          "xyz".getBytes()) == null);
      assertThat(new String(db.get(columnFamilyHandleList.get(1),
          "newcfkey".getBytes()))).isEqualTo("value");
      assertThat(new String(db.get(columnFamilyHandleList.get(1),
          "newcfkey2".getBytes()))).isEqualTo("value2");
      assertThat(new String(db.get("key".getBytes()))).isEqualTo("value");
    } finally {
      if (db != null) {
        db.close();
      }
      if (opt != null) {
        opt.dispose();
      }
    }
  }

  @Test
  public void iteratorOnColumnFamily() throws RocksDBException {
    RocksDB db = null;
    DBOptions options = null;
    RocksIterator rocksIterator = null;
    try {
      options = new DBOptions();
      options.setCreateIfMissing(true);
      options.setCreateMissingColumnFamilies(true);
      List<ColumnFamilyDescriptor> cfNames =
          new ArrayList<>();
      List<ColumnFamilyHandle> columnFamilyHandleList =
          new ArrayList<>();
      cfNames.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
      cfNames.add(new ColumnFamilyDescriptor("new_cf"));

      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(),
          cfNames, columnFamilyHandleList);
      db.put(columnFamilyHandleList.get(1), "newcfkey".getBytes(),
          "value".getBytes());
      db.put(columnFamilyHandleList.get(1), "newcfkey2".getBytes(),
          "value2".getBytes());
      rocksIterator = db.newIterator(
          columnFamilyHandleList.get(1));
      rocksIterator.seekToFirst();
      Map<String, String> refMap = new HashMap<>();
      refMap.put("newcfkey", "value");
      refMap.put("newcfkey2", "value2");
      int i = 0;
      while (rocksIterator.isValid()) {
        i++;
        assertThat(refMap.get(new String(rocksIterator.key()))).
            isEqualTo(new String(rocksIterator.value()));
        rocksIterator.next();
      }
      assertThat(i).isEqualTo(2);
      rocksIterator.dispose();
    } finally {
      if (rocksIterator != null) {
        rocksIterator.dispose();
      }
      if (db != null) {
        db.close();
      }
      if (options != null) {
        options.dispose();
      }
    }
  }

  @Test
  public void multiGet() throws RocksDBException {
    RocksDB db = null;
    DBOptions options = null;
    try {
      options = new DBOptions();
      options.setCreateIfMissing(true);
      options.setCreateMissingColumnFamilies(true);
      List<ColumnFamilyDescriptor> cfDescriptors =
          new ArrayList<>();
      List<ColumnFamilyHandle> columnFamilyHandleList =
          new ArrayList<>();
      cfDescriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
      cfDescriptors.add(new ColumnFamilyDescriptor("new_cf"));

      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(),
          cfDescriptors, columnFamilyHandleList);
      db.put(columnFamilyHandleList.get(0), "key".getBytes(), "value".getBytes());
      db.put(columnFamilyHandleList.get(1), "newcfkey".getBytes(), "value".getBytes());

      List<byte[]> keys = new ArrayList<>();
      keys.add("key".getBytes());
      keys.add("newcfkey".getBytes());
      Map<byte[], byte[]> retValues = db.multiGet(columnFamilyHandleList, keys);
      assertThat(retValues.size()).isEqualTo(2);
      assertThat(new String(retValues.get(keys.get(0))))
          .isEqualTo("value");
      assertThat(new String(retValues.get(keys.get(1))))
          .isEqualTo("value");
      retValues = db.multiGet(new ReadOptions(), columnFamilyHandleList, keys);
      assertThat(retValues.size()).isEqualTo(2);
      assertThat(new String(retValues.get(keys.get(0))))
          .isEqualTo("value");
      assertThat(new String(retValues.get(keys.get(1))))
          .isEqualTo("value");
    } finally {
      if (db != null) {
        db.close();
      }
      if (options != null) {
        options.dispose();
      }
    }
  }

  @Test
  public void properties() throws RocksDBException {
    RocksDB db = null;
    DBOptions options = null;
    try {
      options = new DBOptions();
      options.setCreateIfMissing(true);
      options.setCreateMissingColumnFamilies(true);
      List<ColumnFamilyDescriptor> cfNames =
          new ArrayList<>();
      List<ColumnFamilyHandle> columnFamilyHandleList =
          new ArrayList<>();
      cfNames.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
      cfNames.add(new ColumnFamilyDescriptor("new_cf"));

      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(),
          cfNames, columnFamilyHandleList);
      assertThat(db.getProperty("rocksdb.estimate-num-keys")).
          isNotNull();
      assertThat(db.getLongProperty(columnFamilyHandleList.get(0),
          "rocksdb.estimate-num-keys")).isGreaterThanOrEqualTo(0);
      assertThat(db.getProperty("rocksdb.stats")).isNotNull();
      assertThat(db.getProperty(columnFamilyHandleList.get(0),
          "rocksdb.sstables")).isNotNull();
      assertThat(db.getProperty(columnFamilyHandleList.get(1),
          "rocksdb.estimate-num-keys")).isNotNull();
      assertThat(db.getProperty(columnFamilyHandleList.get(1),
          "rocksdb.stats")).isNotNull();
      assertThat(db.getProperty(columnFamilyHandleList.get(1),
          "rocksdb.sstables")).isNotNull();
    } finally {
      if (db != null) {
        db.close();
      }
      if (options != null) {
        options.dispose();
      }
    }
  }


  @Test
  public void iterators() throws RocksDBException {
    RocksDB db = null;
    DBOptions options = null;
    try {
      options = new DBOptions();
      options.setCreateIfMissing(true);
      options.setCreateMissingColumnFamilies(true);

      List<ColumnFamilyDescriptor> cfNames =
          new ArrayList<>();
      List<ColumnFamilyHandle> columnFamilyHandleList =
          new ArrayList<>();
      cfNames.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
      cfNames.add(new ColumnFamilyDescriptor("new_cf"));

      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(),
          cfNames, columnFamilyHandleList);
      List<RocksIterator> iterators =
          db.newIterators(columnFamilyHandleList);
      assertThat(iterators.size()).isEqualTo(2);
      RocksIterator iter = iterators.get(0);
      iter.seekToFirst();
      Map<String, String> defRefMap = new HashMap<>();
      defRefMap.put("dfkey1", "dfvalue");
      defRefMap.put("key", "value");
      while (iter.isValid()) {
        assertThat(defRefMap.get(new String(iter.key()))).
            isEqualTo(new String(iter.value()));
        iter.next();
      }
      // iterate over new_cf key/value pairs
      Map<String, String> cfRefMap = new HashMap<>();
      cfRefMap.put("newcfkey", "value");
      cfRefMap.put("newcfkey2", "value2");
      iter = iterators.get(1);
      iter.seekToFirst();
      while (iter.isValid()) {
        assertThat(cfRefMap.get(new String(iter.key()))).
            isEqualTo(new String(iter.value()));
        iter.next();
      }
    } finally {
      if (db != null) {
        db.close();
      }
      if (options != null) {
        options.dispose();
      }
    }
  }

  @Test(expected = RocksDBException.class)
  public void failPutDisposedCF() throws RocksDBException {
    RocksDB db = null;
    DBOptions options = null;
    try {
      options = new DBOptions();
      options.setCreateIfMissing(true);
      List<ColumnFamilyDescriptor> cfNames =
          new ArrayList<>();
      List<ColumnFamilyHandle> columnFamilyHandleList =
          new ArrayList<>();
      cfNames.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
      cfNames.add(new ColumnFamilyDescriptor("new_cf"));

      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(),
          cfNames, columnFamilyHandleList);
      db.dropColumnFamily(columnFamilyHandleList.get(1));
      db.put(columnFamilyHandleList.get(1), "key".getBytes(), "value".getBytes());
    } finally {
      if (db != null) {
        db.close();
      }
      if (options != null) {
        options.dispose();
      }
    }
  }

  @Test(expected = RocksDBException.class)
  public void failRemoveDisposedCF() throws RocksDBException {
    RocksDB db = null;
    DBOptions options = null;
    try {
      options = new DBOptions();
      options.setCreateIfMissing(true);
      List<ColumnFamilyDescriptor> cfNames =
          new ArrayList<>();
      List<ColumnFamilyHandle> columnFamilyHandleList =
          new ArrayList<>();
      cfNames.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
      cfNames.add(new ColumnFamilyDescriptor("new_cf"));

      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(),
          cfNames, columnFamilyHandleList);
      db.dropColumnFamily(columnFamilyHandleList.get(1));
      db.remove(columnFamilyHandleList.get(1), "key".getBytes());
    } finally {
      if (db != null) {
        db.close();
      }
      if (options != null) {
        options.dispose();
      }
    }
  }

  @Test(expected = RocksDBException.class)
  public void failGetDisposedCF() throws RocksDBException {
    RocksDB db = null;
    DBOptions options = null;
    try {
      options = new DBOptions();
      options.setCreateIfMissing(true);
      List<ColumnFamilyDescriptor> cfNames =
          new ArrayList<>();
      List<ColumnFamilyHandle> columnFamilyHandleList =
          new ArrayList<>();
      cfNames.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
      cfNames.add(new ColumnFamilyDescriptor("new_cf"));

      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(),
          cfNames, columnFamilyHandleList);
      db.dropColumnFamily(columnFamilyHandleList.get(1));
      db.get(columnFamilyHandleList.get(1), "key".getBytes());
    } finally {
      if (db != null) {
        db.close();
      }
      if (options != null) {
        options.dispose();
      }
    }
  }

  @Test(expected = RocksDBException.class)
  public void failMultiGetWithoutCorrectNumberOfCF() throws RocksDBException {
    RocksDB db = null;
    DBOptions options = null;
    try {
      options = new DBOptions();
      options.setCreateIfMissing(true);
      List<ColumnFamilyDescriptor> cfNames =
          new ArrayList<>();
      List<ColumnFamilyHandle> columnFamilyHandleList =
          new ArrayList<>();
      cfNames.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
      cfNames.add(new ColumnFamilyDescriptor("new_cf"));

      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(),
          cfNames, columnFamilyHandleList);
      List<byte[]> keys = new ArrayList<>();
      keys.add("key".getBytes());
      keys.add("newcfkey".getBytes());
      List<ColumnFamilyHandle> cfCustomList = new ArrayList<>();
      db.multiGet(cfCustomList, keys);

    } finally {
      if (db != null) {
        db.close();
      }
      if (options != null) {
        options.dispose();
      }
    }
  }

}