3114

Build # 3114

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/0671e1352c0860e64baf85512e23d1584fcb6f17">0671e1352">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/0671e1352c0860e64baf85512e23d1584fcb6f17">0671e1352</a><a href="https://github.com/apache/carbondata/commit/0671e1352c0860e64baf85512e23d1584fcb6f17">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/0671e1352c0860e64baf85512e23d1584fcb6f17&quot;&gt;0671e1352&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/0671e1352c0860e64baf85512e23d1584fcb6f17&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a ... (continued)

Run Details

37 of 37 new or added lines in 8 files covered. (100.0%)

62812 of 79634 relevant lines covered (78.88%)

1.04 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.05

/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ForwardDictionaryCache.java

/*
 * 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.carbondata.core.cache.dictionary;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.cache.CacheType;
import org.apache.carbondata.core.cache.CarbonLRUCache;
import org.apache.carbondata.core.reader.CarbonDictionaryColumnMetaChunk;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.ObjectSizeCalculator;
import org.apache.carbondata.core.util.TaskMetricsMap;

import org.apache.log4j.Logger;

/**
 * This class implements methods to create dictionary cache which will hold
 * dictionary chunks for look up of surrogate keys and values
 */
public class ForwardDictionaryCache<K extends
    DictionaryColumnUniqueIdentifier, V extends Dictionary> extends AbstractDictionaryCache<K, V> {

  /**
   * Attribute for Carbon LOGGER
   */
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(ForwardDictionaryCache.class.getName());

  private static final Map<DictionaryColumnUniqueIdentifier, Object> DICTIONARY_LOCK_OBJECT =
      new ConcurrentHashMap<>();

  private static final long sizeOfEmptyDictChunks =
      ObjectSizeCalculator.estimate(new ArrayList<byte[]>(CarbonUtil.getDictionaryChunkSize()), 16);

  private static final long byteArraySize = ObjectSizeCalculator.estimate(new byte[0], 16);

  /**
   * @param carbonLRUCache
   */
  public ForwardDictionaryCache(CarbonLRUCache carbonLRUCache) {
    super(carbonLRUCache);
  }

  /**
   * This method will get the value for the given key. If value does not exist
   * for the given key, it will check and load the value.
   *
   * @param dictionaryColumnUniqueIdentifier unique identifier which contains dbName,
   *                                         tableName and columnIdentifier
   * @return dictionary
   * @throws IOException in case memory is not sufficient to load dictionary into memory
   */
  @Override public Dictionary get(DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier)
      throws IOException {
    return getDictionary(dictionaryColumnUniqueIdentifier);
  }

  /**
   * This method will return a list of values for the given list of keys.
   * For each key, this method will check and load the data if required.
   *
   * @param dictionaryColumnUniqueIdentifiers unique identifier which contains dbName,
   *                                          tableName and columnIdentifier
   * @return list of dictionary
   * @throws IOException in case memory is not sufficient to load dictionary into memory
   */
  @Override public List<Dictionary> getAll(
      List<DictionaryColumnUniqueIdentifier> dictionaryColumnUniqueIdentifiers) throws IOException {
    boolean exceptionOccurredInDictionaryLoading = false;
    String exceptionMessage = "";
    List<Dictionary> forwardDictionaryObjectList =
        new ArrayList<Dictionary>(dictionaryColumnUniqueIdentifiers.size());
    List<Future<Dictionary>> taskSubmitList =
        new ArrayList<>(dictionaryColumnUniqueIdentifiers.size());
    ExecutorService executorService = Executors.newFixedThreadPool(threadPoolSize);
    for (final DictionaryColumnUniqueIdentifier uniqueIdent : dictionaryColumnUniqueIdentifiers) {
      taskSubmitList.add(executorService.submit(new Callable<Dictionary>() {
        @Override public Dictionary call() throws IOException {
          try {
            // Register thread callback for calculating metrics
            TaskMetricsMap.getInstance().registerThreadCallback();
            // in case of multiple task for same query same executor
            // only one task should load the dictionary
            // others will wait on monitor and get the loaded dictionary values
            Object lockObject = DICTIONARY_LOCK_OBJECT.get(uniqueIdent);
            // if lock object is null
            if (null == lockObject) {
              // Acquire the lock on map
              synchronized (DICTIONARY_LOCK_OBJECT) {
                // double checking the dictionary lock object
                lockObject = DICTIONARY_LOCK_OBJECT.get(uniqueIdent);
                // if still it is null add new lock object
                if (null == lockObject) {
                  lockObject = new Object();
                  DICTIONARY_LOCK_OBJECT.put(uniqueIdent, lockObject);
                }
              }
            }
            Dictionary dictionary = null;
            synchronized (lockObject) {
              dictionary = getDictionary(uniqueIdent);
            }
            return dictionary;
          }  finally {
            // update read bytes metrics for this thread
            TaskMetricsMap.getInstance().updateReadBytes(Thread.currentThread().getId());
          }

        }
      }));
    }
    try {
      executorService.shutdown();
      executorService.awaitTermination(2, TimeUnit.HOURS);
    } catch (InterruptedException e) {
      LOGGER.error("Error loading the dictionary: " + e.getMessage(), e);
    }
    for (int i = 0; i < taskSubmitList.size(); i++) {
      try {
        Dictionary columnDictionary = taskSubmitList.get(i).get();
        forwardDictionaryObjectList.add(columnDictionary);
      } catch (Throwable e) {
        exceptionOccurredInDictionaryLoading = true;
        exceptionMessage = e.getMessage();
      }
    }
    if (exceptionOccurredInDictionaryLoading) {
      clearDictionary(forwardDictionaryObjectList);
      LOGGER.error(exceptionMessage);
      throw new IOException(exceptionMessage);
    }
    return forwardDictionaryObjectList;
  }

  /**
   * This method will return the value for the given key. It will not check and load
   * the data for the given key
   *
   * @param dictionaryColumnUniqueIdentifier unique identifier which contains dbName,
   *                                         tableName and columnIdentifier
   * @return
   */
  @Override public Dictionary getIfPresent(
      DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
    Dictionary forwardDictionary = null;
    ColumnDictionaryInfo columnDictionaryInfo = (ColumnDictionaryInfo) carbonLRUCache.get(
        getLruCacheKey(dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId(),
            CacheType.FORWARD_DICTIONARY));
    if (null != columnDictionaryInfo) {
      forwardDictionary = new ForwardDictionary(columnDictionaryInfo);
      incrementDictionaryAccessCount(columnDictionaryInfo);
    }
    return forwardDictionary;
  }

  /**
   * This method will remove the cache for a given key
   *
   * @param dictionaryColumnUniqueIdentifier unique identifier which contains dbName,
   *                                         tableName and columnIdentifier
   */
  @Override public void invalidate(
      DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
    carbonLRUCache.remove(
        getLruCacheKey(dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId(),
            CacheType.FORWARD_DICTIONARY));
  }

  /**
   * This method will get the value for the given key. If value does not exist
   * for the given key, it will check and load the value.
   *
   * @param dictionaryColumnUniqueIdentifier unique identifier which contains dbName,
   *                                         tableName and columnIdentifier
   * @return dictionary
   * @throws IOException in case memory is not sufficient to load dictionary into memory
   */
  private Dictionary getDictionary(
      DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) throws IOException {
    Dictionary forwardDictionary = null;
    // dictionary is only for primitive data type
    assert (!dictionaryColumnUniqueIdentifier.getDataType().isComplexType());
    String columnIdentifier = dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId();
    ColumnDictionaryInfo columnDictionaryInfo =
        getColumnDictionaryInfo(dictionaryColumnUniqueIdentifier, columnIdentifier);
    // load sort index file in case of forward dictionary
    checkAndLoadDictionaryData(dictionaryColumnUniqueIdentifier, columnDictionaryInfo,
        getLruCacheKey(dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId(),
            CacheType.FORWARD_DICTIONARY), true);
    forwardDictionary = new ForwardDictionary(columnDictionaryInfo);
    return forwardDictionary;
  }

  /**
   * This method will check and create columnDictionaryInfo object for the given column
   *
   * @param dictionaryColumnUniqueIdentifier
   * @param columnIdentifier
   * @return
   */
  private ColumnDictionaryInfo getColumnDictionaryInfo(
      DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, String columnIdentifier) {
    ColumnDictionaryInfo columnDictionaryInfo = (ColumnDictionaryInfo) carbonLRUCache
        .get(getLruCacheKey(columnIdentifier, CacheType.FORWARD_DICTIONARY));
    if (null == columnDictionaryInfo) {
      synchronized (dictionaryColumnUniqueIdentifier) {
        columnDictionaryInfo = (ColumnDictionaryInfo) carbonLRUCache
            .get(getLruCacheKey(columnIdentifier, CacheType.FORWARD_DICTIONARY));
        if (null == columnDictionaryInfo) {
          columnDictionaryInfo =
              new ColumnDictionaryInfo(dictionaryColumnUniqueIdentifier.getDataType());
        }
      }
    }
    return columnDictionaryInfo;
  }

  @Override public void clearAccessCount(List<DictionaryColumnUniqueIdentifier> keys) {
    for (DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier : keys) {
      Dictionary cacheable = (Dictionary) carbonLRUCache.get(
          getLruCacheKey(dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId(),
              CacheType.FORWARD_DICTIONARY));
      cacheable.clear();
    }
  }

  @Override protected long getEstimatedDictionarySize(DictionaryInfo dictionaryInfo,
      CarbonDictionaryColumnMetaChunk carbonDictionaryColumnMetaChunk,
      DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, boolean
      readSortIndexSize) throws IOException {
    // required size will be size total size of file - offset till file is
    // already read
    long requiredSize =
        carbonDictionaryColumnMetaChunk.getEnd_offset() -
            dictionaryInfo.getOffsetTillFileIsRead();

    long numOfRecords = dictionaryInfo.getOffsetTillFileIsRead() == 0 ?
        carbonDictionaryColumnMetaChunk.getMax_surrogate_key() :
        carbonDictionaryColumnMetaChunk.getMax_surrogate_key()
            - getNumRecordsInCarbonDictionaryColumnMetaChunk(
            dictionaryColumnUniqueIdentifier,
            dictionaryInfo.getOffsetTillFileIsRead());

    if (numOfRecords > 0) {
      long avgRecordsSize = requiredSize / numOfRecords;
      long bytesPerRecord = (long)Math.ceil(avgRecordsSize / 8.0) * 8;

      requiredSize = (bytesPerRecord + byteArraySize) * numOfRecords;
    }

    if (readSortIndexSize) {
      // every time we are loading all the sort index files.Hence memory calculation for all
      // the records
      requiredSize = requiredSize + getSortIndexSize(
          carbonDictionaryColumnMetaChunk.getMax_surrogate_key());
    }

    return requiredSize + sizeOfEmptyDictChunks;
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.core.cache.dictionary;
19
20	import java.io.IOException;
21	import java.util.ArrayList;
22	import java.util.List;
23	import java.util.Map;
24	import java.util.concurrent.Callable;
25	import java.util.concurrent.ConcurrentHashMap;
26	import java.util.concurrent.ExecutorService;
27	import java.util.concurrent.Executors;
28	import java.util.concurrent.Future;
29	import java.util.concurrent.TimeUnit;
30
31	import org.apache.carbondata.common.logging.LogServiceFactory;
32	import org.apache.carbondata.core.cache.CacheType;
33	import org.apache.carbondata.core.cache.CarbonLRUCache;
34	import org.apache.carbondata.core.reader.CarbonDictionaryColumnMetaChunk;
35	import org.apache.carbondata.core.util.CarbonUtil;
36	import org.apache.carbondata.core.util.ObjectSizeCalculator;
37	import org.apache.carbondata.core.util.TaskMetricsMap;
38
39	import org.apache.log4j.Logger;
40
41	/**
42	* This class implements methods to create dictionary cache which will hold
43	* dictionary chunks for look up of surrogate keys and values
44	*/
45	public class ForwardDictionaryCache<K extends	2✔
46	DictionaryColumnUniqueIdentifier, V extends Dictionary> extends AbstractDictionaryCache<K, V> {
47
48	/**
49	* Attribute for Carbon LOGGER
50	*/
51	private static final Logger LOGGER =	2✔
52	LogServiceFactory.getLogService(ForwardDictionaryCache.class.getName());	2✔
53
54	private static final Map<DictionaryColumnUniqueIdentifier, Object> DICTIONARY_LOCK_OBJECT =	2✔
55	new ConcurrentHashMap<>();
56
57	private static final long sizeOfEmptyDictChunks =	2✔
58	ObjectSizeCalculator.estimate(new ArrayList<byte[]>(CarbonUtil.getDictionaryChunkSize()), 16);	2✔
59
60	private static final long byteArraySize = ObjectSizeCalculator.estimate(new byte[0], 16);	2✔
61
62	/**
63	* @param carbonLRUCache
64	*/
65	public ForwardDictionaryCache(CarbonLRUCache carbonLRUCache) {
66	super(carbonLRUCache);	2✔
67	}	2✔
68
69	/**
70	* This method will get the value for the given key. If value does not exist
71	* for the given key, it will check and load the value.
72	*
73	* @param dictionaryColumnUniqueIdentifier unique identifier which contains dbName,
74	* tableName and columnIdentifier
75	* @return dictionary
76	* @throws IOException in case memory is not sufficient to load dictionary into memory
77	*/
78	@Override public Dictionary get(DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier)
79	throws IOException {
80	return getDictionary(dictionaryColumnUniqueIdentifier);	2✔
81	}
82
83	/**
84	* This method will return a list of values for the given list of keys.
85	* For each key, this method will check and load the data if required.
86	*
87	* @param dictionaryColumnUniqueIdentifiers unique identifier which contains dbName,
88	* tableName and columnIdentifier
89	* @return list of dictionary
90	* @throws IOException in case memory is not sufficient to load dictionary into memory
91	*/
92	@Override public List<Dictionary> getAll(
93	List<DictionaryColumnUniqueIdentifier> dictionaryColumnUniqueIdentifiers) throws IOException {
94	boolean exceptionOccurredInDictionaryLoading = false;	2✔
95	String exceptionMessage = "";	2✔
96	List<Dictionary> forwardDictionaryObjectList =	2✔
97	new ArrayList<Dictionary>(dictionaryColumnUniqueIdentifiers.size());	2✔
98	List<Future<Dictionary>> taskSubmitList =	2✔
99	new ArrayList<>(dictionaryColumnUniqueIdentifiers.size());	2✔
100	ExecutorService executorService = Executors.newFixedThreadPool(threadPoolSize);	2✔
101	for (final DictionaryColumnUniqueIdentifier uniqueIdent : dictionaryColumnUniqueIdentifiers) {	2✔
102	taskSubmitList.add(executorService.submit(new Callable<Dictionary>() {	2✔
103	@Override public Dictionary call() throws IOException {
104	try {
105	// Register thread callback for calculating metrics
106	TaskMetricsMap.getInstance().registerThreadCallback();	2✔
107	// in case of multiple task for same query same executor
108	// only one task should load the dictionary
109	// others will wait on monitor and get the loaded dictionary values
110	Object lockObject = DICTIONARY_LOCK_OBJECT.get(uniqueIdent);	2✔
111	// if lock object is null
112	if (null == lockObject) {	2✔
113	// Acquire the lock on map
114	synchronized (DICTIONARY_LOCK_OBJECT) {	2✔
115	// double checking the dictionary lock object
116	lockObject = DICTIONARY_LOCK_OBJECT.get(uniqueIdent);	2✔
117	// if still it is null add new lock object
118	if (null == lockObject) {	2✔
119	lockObject = new Object();	2✔
120	DICTIONARY_LOCK_OBJECT.put(uniqueIdent, lockObject);	2✔
121	}
122	}	2✔
123	}
124	Dictionary dictionary = null;	2✔
125	synchronized (lockObject) {	2✔
126	dictionary = getDictionary(uniqueIdent);	2✔
127	}	2✔
128	return dictionary;	2✔
129	} finally {
130	// update read bytes metrics for this thread
131	TaskMetricsMap.getInstance().updateReadBytes(Thread.currentThread().getId());	2✔
132	}	×
133
134	}
135	}));
136	}	2✔
137	try {
138	executorService.shutdown();	2✔
139	executorService.awaitTermination(2, TimeUnit.HOURS);	2✔
140	} catch (InterruptedException e) {	×
141	LOGGER.error("Error loading the dictionary: " + e.getMessage(), e);	×
142	}	2✔
143	for (int i = 0; i < taskSubmitList.size(); i++) {	2✔
144	try {
145	Dictionary columnDictionary = taskSubmitList.get(i).get();	2✔
146	forwardDictionaryObjectList.add(columnDictionary);	2✔
147	} catch (Throwable e) {	×
148	exceptionOccurredInDictionaryLoading = true;	×
149	exceptionMessage = e.getMessage();	×
150	}	2✔
151	}
152	if (exceptionOccurredInDictionaryLoading) {	2✔
153	clearDictionary(forwardDictionaryObjectList);	×
154	LOGGER.error(exceptionMessage);	×
155	throw new IOException(exceptionMessage);	×
156	}
157	return forwardDictionaryObjectList;	2✔
158	}
159
160	/**
161	* This method will return the value for the given key. It will not check and load
162	* the data for the given key
163	*
164	* @param dictionaryColumnUniqueIdentifier unique identifier which contains dbName,
165	* tableName and columnIdentifier
166	* @return
167	*/
168	@Override public Dictionary getIfPresent(
169	DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
170	Dictionary forwardDictionary = null;	×
171	ColumnDictionaryInfo columnDictionaryInfo = (ColumnDictionaryInfo) carbonLRUCache.get(	×
172	getLruCacheKey(dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId(),	×
173	CacheType.FORWARD_DICTIONARY));
174	if (null != columnDictionaryInfo) {	×
175	forwardDictionary = new ForwardDictionary(columnDictionaryInfo);	×
176	incrementDictionaryAccessCount(columnDictionaryInfo);	×
177	}
178	return forwardDictionary;	×
179	}
180
181	/**
182	* This method will remove the cache for a given key
183	*
184	* @param dictionaryColumnUniqueIdentifier unique identifier which contains dbName,
185	* tableName and columnIdentifier
186	*/
187	@Override public void invalidate(
188	DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
189	carbonLRUCache.remove(	2✔
190	getLruCacheKey(dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId(),	2✔
191	CacheType.FORWARD_DICTIONARY));
192	}	2✔
193
194	/**
195	* This method will get the value for the given key. If value does not exist
196	* for the given key, it will check and load the value.
197	*
198	* @param dictionaryColumnUniqueIdentifier unique identifier which contains dbName,
199	* tableName and columnIdentifier
200	* @return dictionary
201	* @throws IOException in case memory is not sufficient to load dictionary into memory
202	*/
203	private Dictionary getDictionary(
204	DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) throws IOException {
205	Dictionary forwardDictionary = null;	2✔
206	// dictionary is only for primitive data type
207	assert (!dictionaryColumnUniqueIdentifier.getDataType().isComplexType());	2✔
208	String columnIdentifier = dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId();	2✔
209	ColumnDictionaryInfo columnDictionaryInfo =	2✔
210	getColumnDictionaryInfo(dictionaryColumnUniqueIdentifier, columnIdentifier);	2✔
211	// load sort index file in case of forward dictionary
212	checkAndLoadDictionaryData(dictionaryColumnUniqueIdentifier, columnDictionaryInfo,	2✔
213	getLruCacheKey(dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId(),	2✔
214	CacheType.FORWARD_DICTIONARY), true);
215	forwardDictionary = new ForwardDictionary(columnDictionaryInfo);	2✔
216	return forwardDictionary;	2✔
217	}
218
219	/**
220	* This method will check and create columnDictionaryInfo object for the given column
221	*
222	* @param dictionaryColumnUniqueIdentifier
223	* @param columnIdentifier
224	* @return
225	*/
226	private ColumnDictionaryInfo getColumnDictionaryInfo(
227	DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, String columnIdentifier) {
228	ColumnDictionaryInfo columnDictionaryInfo = (ColumnDictionaryInfo) carbonLRUCache	2✔
229	.get(getLruCacheKey(columnIdentifier, CacheType.FORWARD_DICTIONARY));	2✔
230	if (null == columnDictionaryInfo) {	2✔
231	synchronized (dictionaryColumnUniqueIdentifier) {	2✔
232	columnDictionaryInfo = (ColumnDictionaryInfo) carbonLRUCache	2✔
233	.get(getLruCacheKey(columnIdentifier, CacheType.FORWARD_DICTIONARY));	2✔
234	if (null == columnDictionaryInfo) {	2✔
235	columnDictionaryInfo =	2✔
236	new ColumnDictionaryInfo(dictionaryColumnUniqueIdentifier.getDataType());	2✔
237	}
238	}	2✔
239	}
240	return columnDictionaryInfo;	2✔
241	}
242
243	@Override public void clearAccessCount(List<DictionaryColumnUniqueIdentifier> keys) {
244	for (DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier : keys) {	×
245	Dictionary cacheable = (Dictionary) carbonLRUCache.get(	×
246	getLruCacheKey(dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId(),	×
247	CacheType.FORWARD_DICTIONARY));
248	cacheable.clear();	×
249	}	×
250	}	×
251
252	@Override protected long getEstimatedDictionarySize(DictionaryInfo dictionaryInfo,
253	CarbonDictionaryColumnMetaChunk carbonDictionaryColumnMetaChunk,
254	DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier, boolean
255	readSortIndexSize) throws IOException {
256	// required size will be size total size of file - offset till file is
257	// already read
258	long requiredSize =	2✔
259	carbonDictionaryColumnMetaChunk.getEnd_offset() -	2✔
260	dictionaryInfo.getOffsetTillFileIsRead();	2✔
261
262	long numOfRecords = dictionaryInfo.getOffsetTillFileIsRead() == 0 ?	2✔
263	carbonDictionaryColumnMetaChunk.getMax_surrogate_key() :	2✔
264	carbonDictionaryColumnMetaChunk.getMax_surrogate_key()	2✔
265	- getNumRecordsInCarbonDictionaryColumnMetaChunk(	2✔
266	dictionaryColumnUniqueIdentifier,
267	dictionaryInfo.getOffsetTillFileIsRead());	2✔
268
269	if (numOfRecords > 0) {	2✔
270	long avgRecordsSize = requiredSize / numOfRecords;	2✔
271	long bytesPerRecord = (long)Math.ceil(avgRecordsSize / 8.0) * 8;	2✔
272
273	requiredSize = (bytesPerRecord + byteArraySize) * numOfRecords;	2✔
274	}
275
276	if (readSortIndexSize) {	2✔
277	// every time we are loading all the sort index files.Hence memory calculation for all
278	// the records
279	requiredSize = requiredSize + getSortIndexSize(	2✔
280	carbonDictionaryColumnMetaChunk.getMax_surrogate_key());	2✔
281	}
282
283	return requiredSize + sizeOfEmptyDictChunks;	2✔
284	}
285	}

apache / carbondata / 3114

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous