From bf0b9d1e4d1c38e06721bd169e3cdb6782701a9d Mon Sep 17 00:00:00 2001
From: kamijin_fanta <kamijin@live.jp>
Date: Wed, 18 Oct 2017 14:25:56 +0900
Subject: [PATCH] Add 'use_ttl' new option. disable unnecessary reading at
 compaction.

---
 doc/documentation.rst                             |  3 +++
 .../stratio/cassandra/lucene/IndexOptions.scala   | 15 ++++++++++-----
 .../stratio/cassandra/lucene/IndexWriter.scala    |  3 ++-
 3 files changed, 15 insertions(+), 6 deletions(-)

diff --git a/doc/documentation.rst b/doc/documentation.rst
index 71089d891..10dd8e921 100644
--- a/doc/documentation.rst
+++ b/doc/documentation.rst
@@ -583,6 +583,9 @@ All options take a value enclosed in single quotes:
    The cost of this optimization is an extra comparison performed each time a row must be indexed.
    This flag helps in reducing lucene calls when the row is updated partially, and the columns
    that affect the index are updated less frequently then the rest of the row.
+-  **use_ttl**: If false, deletion of data using `USING TTL` will not be performed automatically.
+   This reduces reading of data during compaction.
+   In some cases, the speed of compaction is greatly improved.
 -  **schema**: see below
 
 .. code-block:: sql
diff --git a/plugin/src/main/scala/com/stratio/cassandra/lucene/IndexOptions.scala b/plugin/src/main/scala/com/stratio/cassandra/lucene/IndexOptions.scala
index b6a6470e2..e411dac13 100644
--- a/plugin/src/main/scala/com/stratio/cassandra/lucene/IndexOptions.scala
+++ b/plugin/src/main/scala/com/stratio/cassandra/lucene/IndexOptions.scala
@@ -69,7 +69,9 @@ class IndexOptions(tableMetadata: CFMetaData, indexMetadata: IndexMetadata) {
   val path = parsePath(options, tableMetadata, Some(indexMetadata))
 
   /** If the index is sparse or not */
-  val sparse = parseSparse(options, tableMetadata)
+  val sparse = parseBool(options, SPARSE_OPTION, DEFAULT_SPARSE)
+
+  val useTtl = parseBool(options, USE_TTL, DEFAULT_USE_TTL)
 }
 
 /** Companion object for [[IndexOptions]]. */
@@ -107,6 +109,9 @@ object IndexOptions {
   val SPARSE_OPTION = "sparse"
   val DEFAULT_SPARSE = false
 
+  val USE_TTL = "use_ttl"
+  val DEFAULT_USE_TTL = true
+
   /** Validates the specified index options.
     *
     * @param options  the options to be validated
@@ -192,12 +197,12 @@ object IndexOptions {
       }).getOrElse(DEFAULT_PARTITIONER)
   }
 
-  def parseSparse(options: Map[String, String], table: CFMetaData): Boolean = {
-    options.get(SPARSE_OPTION).map(
+  def parseBool(options: Map[String, String], name: String, default: Boolean): Boolean = {
+    options.get(name).map(
       value => try value.toBoolean catch {
         case e: Exception => throw new IndexException(e,
-          s"'$SPARSE_OPTION' is invalid : ${e.getMessage}")
-      }).getOrElse(DEFAULT_SPARSE)
+          s"'$name' is invalid : ${e.getMessage}")
+      }).getOrElse(default)
   }
 
   private def parseInt(options: Map[String, String], name: String, default: Int): Int = {
diff --git a/plugin/src/main/scala/com/stratio/cassandra/lucene/IndexWriter.scala b/plugin/src/main/scala/com/stratio/cassandra/lucene/IndexWriter.scala
index a6a917766..701fe7d82 100644
--- a/plugin/src/main/scala/com/stratio/cassandra/lucene/IndexWriter.scala
+++ b/plugin/src/main/scala/com/stratio/cassandra/lucene/IndexWriter.scala
@@ -20,7 +20,7 @@ import org.apache.cassandra.db._
 import org.apache.cassandra.db.rows.{Row, RowIterator, UnfilteredRowIterators}
 import org.apache.cassandra.index.Index.Indexer
 import org.apache.cassandra.index.transactions.IndexTransaction
-import org.apache.cassandra.index.transactions.IndexTransaction.Type.CLEANUP
+import org.apache.cassandra.index.transactions.IndexTransaction.Type.{CLEANUP, COMPACTION}
 import org.apache.cassandra.utils.concurrent.OpOrder
 
 /** [[Indexer]] for Lucene-based index.
@@ -116,6 +116,7 @@ abstract class IndexWriter(
 
     // Skip on cleanups
     if (transactionType == CLEANUP) return
+    if (transactionType == COMPACTION && !service.options.useTtl) return
 
     // Finish with mutual exclusion on partition
     service.readBeforeWriteLocker.run(key, () => commit())