/*
 * 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.kylin.storage.hbase.cube.v2.filter;

import java.io.IOException;
import java.io.InputStream;
import java.util.List;
import java.util.Set;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.Pair;
import org.apache.kylin.dimension.DimensionEncoding;
import org.apache.kylin.metadata.filter.UDF.MassInValueProvider;
import org.apache.kylin.metadata.filter.function.Functions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Stopwatch;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.RemovalListener;
import com.google.common.cache.RemovalNotification;
import com.google.common.collect.Sets;

public class MassInValueProviderImpl implements MassInValueProvider {
    public static final Logger logger = LoggerFactory.getLogger(MassInValueProviderImpl.class);

    private final static Cache<String, Pair<Long, Set<ByteArray>>> hdfs_caches = CacheBuilder.newBuilder().maximumSize(10).weakValues().removalListener(new RemovalListener<Object, Object>() {
        @Override
        public void onRemoval(RemovalNotification<Object, Object> notification) {
            logger.debug(String.valueOf(notification.getCause()));
        }
    }).build();

    private Set<ByteArray> ret = Sets.newHashSet();

    public MassInValueProviderImpl(Functions.FilterTableType filterTableType, String filterResourceIdentifier, DimensionEncoding encoding) {

        if (filterTableType == Functions.FilterTableType.HDFS) {

            logger.info("Start to load HDFS filter table from " + filterResourceIdentifier);
            Stopwatch stopwatch = new Stopwatch().start();

            FileSystem fileSystem = null;
            try {
                fileSystem = FileSystem.get(HBaseConfiguration.create());

                long modificationTime = fileSystem.getFileStatus(new Path(filterResourceIdentifier)).getModificationTime();
                Pair<Long, Set<ByteArray>> cached = hdfs_caches.getIfPresent(filterResourceIdentifier);
                if (cached != null && cached.getFirst().equals(modificationTime)) {
                    ret = cached.getSecond();
                    logger.info("Load HDFS from cache using " + stopwatch.elapsedMillis() + " millis");
                    return;
                }

                InputStream inputStream = fileSystem.open(new Path(filterResourceIdentifier));
                List<String> lines = IOUtils.readLines(inputStream);

                logger.info("Load HDFS finished after " + stopwatch.elapsedMillis() + " millis");

                for (String line : lines) {
                    if (StringUtils.isEmpty(line)) {
                        continue;
                    }

                    try {
                        ByteArray byteArray = ByteArray.allocate(encoding.getLengthOfEncoding());
                        encoding.encode(line.getBytes(), line.getBytes().length, byteArray.array(), 0);
                        ret.add(byteArray);
                    } catch (Exception e) {
                        logger.warn("Error when encoding the filter line " + line);
                    }
                }

                hdfs_caches.put(filterResourceIdentifier, Pair.newPair(modificationTime, ret));

                logger.info("Mass In values constructed after " + stopwatch.elapsedMillis() + " millis, containing " + ret.size() + " entries");

            } catch (IOException e) {
                throw new RuntimeException("error when loading the mass in values", e);
            }
        } else {
            throw new RuntimeException("HBASE_TABLE FilterTableType Not supported yet");
        }
    }

    @Override
    public Set<?> getMassInValues() {
        return ret;
    }
}

@@ -2048,23 +2048,9 @@
 ize(
-10).weakValues(
+3
 ).re
@@ -2763,32 +2763,102 @@
           try %7B%0A
+                synchronized (hdfs_caches) %7B%0A                    %0A    
                 
@@ -2912,24 +2912,28 @@
 create());%0A%0A
+    
             
@@ -3060,16 +3060,20 @@
         
+ 
+   
 Pair%3CLon
@@ -3144,32 +3144,36 @@
 rceIdentifier);%0A
+    
                 
@@ -3260,16 +3260,20 @@
         
+    
 ret = ca
@@ -3302,32 +3302,36 @@
                 
+  
+  
 logger.info(%22Loa
@@ -3414,24 +3414,28 @@
             
+    
 return;%0A    
@@ -3438,32 +3438,36 @@
                 
+    
 %7D%0A%0A             
@@ -3469,16 +3469,20 @@
         
+  
+  
 InputStr
@@ -3568,16 +3568,20 @@
         
+    
 List%3CStr
@@ -3634,32 +3634,36 @@
                 
+ 
+   
 logger.info(%22Loa
@@ -3745,16 +3745,20 @@
         
+   
+ 
 for (Str
@@ -3773,24 +3773,28 @@
  : lines) %7B%0A
+    
             
@@ -3858,16 +3858,20 @@
         
+    
 continue
@@ -3884,35 +3884,43 @@
                 
+   
+ 
 %7D%0A%0A
+    
                 
@@ -3921,32 +3921,36 @@
           try %7B%0A
+    
                 
@@ -4027,24 +4027,28 @@
 ncoding());%0A
+    
             
@@ -4163,16 +4163,20 @@
         
+    
 ret.add(
@@ -4187,16 +4187,20 @@
 Array);%0A
+    
         
@@ -4227,32 +4227,36 @@
 (Exception e) %7B%0A
+    
                 
@@ -4335,34 +4335,42 @@
                 
-%7D%0A
+    %7D%0A    
                 
@@ -4364,32 +4364,36 @@
              %7D%0A%0A
+    
                 
@@ -4469,24 +4469,28 @@
 me, ret));%0A%0A
+    
             
@@ -4618,24 +4618,58 @@
  entries%22);%0A
+                %7D%0A                
 %0A           

revise hdfs filter cache