PHOENIX-5501 Add support for VIEW_TTL table property during DDL.

Signed-off-by: Chinmay Kulkarni <chinmayskulkarni@apache.org>

phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java

@@ -72,6 +72,10 @@
 import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_STATEMENT_BYTES;
 import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_TYPE_BYTES;
 import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_INDEX_ID_DATA_TYPE_BYTES;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_TTL_BYTES;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_TTL_HWM_BYTES;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_TTL_NOT_DEFINED;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MIN_VIEW_TTL_HWM;
 import static org.apache.phoenix.query.QueryConstants.VIEW_MODIFIED_PROPERTY_TAG_TYPE;
 import static org.apache.phoenix.schema.PTableType.INDEX;
 import static org.apache.phoenix.util.SchemaUtil.getVarCharLength;
@@ -318,6 +322,8 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements Coprocesso
     private static final KeyValue STORAGE_SCHEME_KV = createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, STORAGE_SCHEME_BYTES);
     private static final KeyValue ENCODING_SCHEME_KV = createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, ENCODING_SCHEME_BYTES);
     private static final KeyValue USE_STATS_FOR_PARALLELIZATION_KV = createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, USE_STATS_FOR_PARALLELIZATION_BYTES);
+    private static final KeyValue VIEW_TTL_KV = createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, VIEW_TTL_BYTES);
+    private static final KeyValue VIEW_TTL_HWM_KV = createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, VIEW_TTL_HWM_BYTES);
 
     private static final List<KeyValue> TABLE_KV_COLUMNS = Arrays.<KeyValue>asList(
             EMPTY_KEYVALUE_KV,
@@ -349,7 +355,9 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements Coprocesso
             APPEND_ONLY_SCHEMA_KV,
             STORAGE_SCHEME_KV,
             ENCODING_SCHEME_KV,
-            USE_STATS_FOR_PARALLELIZATION_KV
+            USE_STATS_FOR_PARALLELIZATION_KV,
+            VIEW_TTL_KV,
+            VIEW_TTL_HWM_KV
     );
 
     static {
@@ -385,6 +393,8 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements Coprocesso
     private static final int STORAGE_SCHEME_INDEX = TABLE_KV_COLUMNS.indexOf(STORAGE_SCHEME_KV);
     private static final int QUALIFIER_ENCODING_SCHEME_INDEX = TABLE_KV_COLUMNS.indexOf(ENCODING_SCHEME_KV);
     private static final int USE_STATS_FOR_PARALLELIZATION_INDEX = TABLE_KV_COLUMNS.indexOf(USE_STATS_FOR_PARALLELIZATION_KV);
+    private static final int VIEW_TTL_INDEX = TABLE_KV_COLUMNS.indexOf(VIEW_TTL_KV);
+    private static final int VIEW_TTL_HWM_INDEX = TABLE_KV_COLUMNS.indexOf(VIEW_TTL_HWM_KV);
 
     // KeyValues for Column
     private static final KeyValue DECIMAL_DIGITS_KV = createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, DECIMAL_DIGITS_BYTES);
@@ -1121,6 +1131,22 @@ private PTable getTable(RegionScanner scanner, long clientTimeStamp, long tableT
         Cell useStatsForParallelizationKv = tableKeyValues[USE_STATS_FOR_PARALLELIZATION_INDEX];
         Boolean useStatsForParallelization = useStatsForParallelizationKv == null ? null : Boolean.TRUE.equals(PBoolean.INSTANCE.toObject(useStatsForParallelizationKv.getValueArray(), useStatsForParallelizationKv.getValueOffset(), useStatsForParallelizationKv.getValueLength()));
 
+        Cell viewTTLKv = tableKeyValues[VIEW_TTL_INDEX];
+        long viewTTL = viewTTLKv == null ? VIEW_TTL_NOT_DEFINED :
+                PLong.INSTANCE.getCodec().decodeLong(viewTTLKv.getValueArray(),
+                        viewTTLKv.getValueOffset(), SortOrder.getDefault());
+
+        Cell viewTTLHWMKv = tableKeyValues[VIEW_TTL_HWM_INDEX];
+        long viewTTLHWM = viewTTLHWMKv == null ? MIN_VIEW_TTL_HWM :
+                PLong.INSTANCE.getCodec().decodeLong(viewTTLHWMKv.getValueArray(),
+                        viewTTLHWMKv.getValueOffset(), SortOrder.getDefault());
+
+        // Check the cell tag to see whether the view has modified this property
+        final byte[] tagViewTTL = (viewTTLKv == null) ?
+                HConstants.EMPTY_BYTE_ARRAY : CellUtil.getTagArray(viewTTLKv);
+        boolean viewModifiedViewTTL = (PTableType.VIEW.equals(tableType)) &&
+                Bytes.contains(tagViewTTL, VIEW_MODIFIED_PROPERTY_BYTES);
+
         // Check the cell tag to see whether the view has modified this property
         final byte[] tagUseStatsForParallelization = (useStatsForParallelizationKv == null) ?
                 HConstants.EMPTY_BYTE_ARRAY : CellUtil.getTagArray(useStatsForParallelizationKv);
@@ -1196,6 +1222,8 @@ private PTable getTable(RegionScanner scanner, long clientTimeStamp, long tableT
                 .setBaseColumnCount(baseColumnCount)
                 .setEncodedCQCounter(cqCounter)
                 .setUseStatsForParallelization(useStatsForParallelization)
+                .setViewTTL(viewTTL)
+                .setViewTTLHighWaterMark(viewTTLHWM)
                 .setExcludedColumns(ImmutableList.<PColumn>of())
                 .setTenantId(tenantId)
                 .setSchemaName(schemaName)
@@ -1211,6 +1239,7 @@ private PTable getTable(RegionScanner scanner, long clientTimeStamp, long tableT
                         ImmutableList.<PName>of() : ImmutableList.copyOf(physicalTables))
                 .setViewModifiedUpdateCacheFrequency(viewModifiedUpdateCacheFrequency)
                 .setViewModifiedUseStatsForParallelization(viewModifiedUseStatsForParallelization)
+                .setViewModifiedViewTTL(viewModifiedViewTTL)
                 .setColumns(columns)
                 .build();
     }