2204

Build # 2204

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/671fa03a32077809e1a206f7d97b1468b801d298">671fa03a3<a href="https://github.com/apache/carbondata/commit/671fa03a32077809e1a206f7d97b1468b801d298">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/69220fce0a4e2196e0a094ac3a0f5c5fb4013bd5">69220fce0</a>">69220fce0</a><a href="https://github.com/apache/carbondata/commit/671fa03a32077809e1a206f7d97b1468b801d298"> into 04b52568d">04b52568d</a>

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

61513 of 77934 relevant lines covered (78.93%)

1.05 hits per line

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

84.48

/core/src/main/java/org/apache/carbondata/core/datastore/blocklet/BlockletEncodedColumnPage.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.datastore.blocklet;

import java.io.IOException;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.datastore.page.ActualDataBasedFallbackEncoder;
import org.apache.carbondata.core.datastore.page.DecoderBasedFallbackEncoder;
import org.apache.carbondata.core.datastore.page.FallbackEncodedColumnPage;
import org.apache.carbondata.core.datastore.page.encoding.EncodedColumnPage;
import org.apache.carbondata.core.localdictionary.PageLevelDictionary;
import org.apache.carbondata.core.localdictionary.generator.LocalDictionaryGenerator;
import org.apache.carbondata.core.memory.MemoryException;
import org.apache.carbondata.format.LocalDictionaryChunk;

import org.apache.log4j.Logger;

/**
 * Maintains the list of encoded page of a column in a blocklet
 * and encoded dictionary values only if column is encoded using local
 * dictionary
 * Handle the fallback if all the pages in blocklet are not
 * encoded with local dictionary
 */
public class BlockletEncodedColumnPage {

  /**
   * LOGGER
   */
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(BlockletEncodedColumnPage.class.getName());

  /**
   * list of encoded page of a column in a blocklet
   */
  private List<EncodedColumnPage> encodedColumnPageList;

  /**
   * fallback executor service
   */
  private ExecutorService fallbackExecutorService;

  /**
   * to check whether pages are local dictionary encoded or not
   */
  private boolean isLocalDictEncoded;

  /**
   * page level dictionary only when column is encoded with local dictionary
   */
  private PageLevelDictionary pageLevelDictionary;

  /**
   * fallback future task queue;
   */
  private ArrayDeque<Future<FallbackEncodedColumnPage>> fallbackFutureQueue;

  private String columnName;

  /**
   * is decoder based fallback enabled
   */
  private boolean isDecoderBasedFallBackEnabled;

  /**
   * Local dictionary generator for column
   */
  private LocalDictionaryGenerator localDictionaryGenerator;

  BlockletEncodedColumnPage(ExecutorService fallbackExecutorService,
      boolean isDecoderBasedFallBackEnabled, LocalDictionaryGenerator localDictionaryGenerator) {
    this.fallbackExecutorService = fallbackExecutorService;
    this.fallbackFutureQueue = new ArrayDeque<>();
    this.isDecoderBasedFallBackEnabled = isDecoderBasedFallBackEnabled;
    this.localDictionaryGenerator = localDictionaryGenerator;
  }

  /**
   * Below method will be used to add column page of a column
   *
   * @param encodedColumnPage
   * encoded column page
   */
  void addEncodedColumnPage(EncodedColumnPage encodedColumnPage) {
    if (null == encodedColumnPageList) {
      this.encodedColumnPageList = new ArrayList<>();
      // if dimension page is local dictionary enabled and encoded with local dictionary
      if (encodedColumnPage.isLocalDictGeneratedPage()) {
        this.isLocalDictEncoded = true;
        // get first page dictionary
        this.pageLevelDictionary = encodedColumnPage.getPageDictionary();
      }
      this.encodedColumnPageList.add(encodedColumnPage);
      this.columnName = encodedColumnPage.getActualPage().getColumnSpec().getFieldName();
      return;
    }
    // when first page was encoded without dictionary and next page encoded with dictionary
    // in a blocklet
    if (!isLocalDictEncoded && encodedColumnPage.isLocalDictGeneratedPage()) {
      LOGGER.info(
          "Local dictionary Fallback is initiated for column: " + this.columnName + " for page:"
              + encodedColumnPageList.size());
      initiateFallBack(encodedColumnPage, encodedColumnPageList.size());
      // fill null so once page is decoded again fill the re-encoded page again
      this.encodedColumnPageList.add(null);
    }
    // if local dictionary is false or column is encoded with local dictionary then
    // add a page
    else if (!isLocalDictEncoded || encodedColumnPage.isLocalDictGeneratedPage()) {
      // merge page level dictionary values
      if (null != this.pageLevelDictionary) {
        pageLevelDictionary.mergerDictionaryValues(encodedColumnPage.getPageDictionary());
      }
      this.encodedColumnPageList.add(encodedColumnPage);
    }
    // if all the older pages were encoded with dictionary and new pages are without dictionary
    else {
      isLocalDictEncoded = false;
      pageLevelDictionary = null;
      LOGGER.info("Local dictionary Fallback is initiated for column: " + this.columnName
          + " for pages: 1 to " + encodedColumnPageList.size());
      // submit all the older pages encoded with dictionary for fallback
      for (int pageIndex = 0; pageIndex < encodedColumnPageList.size(); pageIndex++) {
        if (encodedColumnPageList.get(pageIndex).getActualPage().isLocalDictGeneratedPage()) {
          initiateFallBack(encodedColumnPageList.get(pageIndex), pageIndex);
        }
      }
      //add to page list
      this.encodedColumnPageList.add(encodedColumnPage);
    }
  }

  /**
   * Return the list of encoded page list for a column in a blocklet
   *
   * @return list of encoded page list
   */
  public List<EncodedColumnPage> getEncodedColumnPageList() {
    // if fallback queue is null then for some pages fallback was initiated
    if (null != this.fallbackFutureQueue) {
      try {
        // check if queue is not empty
        while (!fallbackFutureQueue.isEmpty()) {
          // get the head element of queue
          FallbackEncodedColumnPage fallbackEncodedColumnPage = fallbackFutureQueue.poll().get();
          // add the encoded column page to list
          encodedColumnPageList.set(fallbackEncodedColumnPage.getPageIndex(),
              fallbackEncodedColumnPage.getEncodedColumnPage());
        }
      } catch (ExecutionException | InterruptedException e) {
        throw new RuntimeException("Problem while encoding the blocklet data during fallback", e);
      }
      // setting to null as all the fallback encoded page has been added to list
      fallbackFutureQueue = null;
    }
    // in case of dictionary encoded column page memory will be freed only after
    // all the pages are added in a blocklet, as fallback can happen anytime so old pages memory
    // cannot be freed, so after encoding is done we can free the page memory
    if (null != pageLevelDictionary) {
      // clear the memory footprint for local dictionary encoded pages
      for (EncodedColumnPage columnPage : encodedColumnPageList) {
        columnPage.freeMemory();
      }
    }
    return encodedColumnPageList;
  }

  /**
   * Below method will be used to get the encoded dictionary
   * values for local dictionary generated columns
   *
   * @return encoded dictionary values if column is local dictionary generated
   */
  public LocalDictionaryChunk getEncodedDictionary() {
    if (null != pageLevelDictionary) {
      try {
        return pageLevelDictionary.getLocalDictionaryChunkForBlocklet();
      } catch (IOException | MemoryException e) {
        throw new RuntimeException(e);
      }
    }
    return null;
  }

  /**
   * This method initiates the fallback for local dictionary encoded column page
   * @param encodedColumnPage
   * @param pageIndex
   */
  private void initiateFallBack(EncodedColumnPage encodedColumnPage, int pageIndex) {
    if (isDecoderBasedFallBackEnabled) {
      fallbackFutureQueue.add(fallbackExecutorService.submit(
          new DecoderBasedFallbackEncoder(encodedColumnPage, pageIndex, localDictionaryGenerator)));
    } else {
      fallbackFutureQueue.add(fallbackExecutorService.submit(
          new ActualDataBasedFallbackEncoder(encodedColumnPage, pageIndex)));
    }
  }
}

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	package org.apache.carbondata.core.datastore.blocklet;
18
19	import java.io.IOException;
20	import java.util.ArrayDeque;
21	import java.util.ArrayList;
22	import java.util.List;
23	import java.util.concurrent.ExecutionException;
24	import java.util.concurrent.ExecutorService;
25	import java.util.concurrent.Future;
26
27	import org.apache.carbondata.common.logging.LogServiceFactory;
28	import org.apache.carbondata.core.datastore.page.ActualDataBasedFallbackEncoder;
29	import org.apache.carbondata.core.datastore.page.DecoderBasedFallbackEncoder;
30	import org.apache.carbondata.core.datastore.page.FallbackEncodedColumnPage;
31	import org.apache.carbondata.core.datastore.page.encoding.EncodedColumnPage;
32	import org.apache.carbondata.core.localdictionary.PageLevelDictionary;
33	import org.apache.carbondata.core.localdictionary.generator.LocalDictionaryGenerator;
34	import org.apache.carbondata.core.memory.MemoryException;
35	import org.apache.carbondata.format.LocalDictionaryChunk;
36
37	import org.apache.log4j.Logger;
38
39	/**
40	* Maintains the list of encoded page of a column in a blocklet
41	* and encoded dictionary values only if column is encoded using local
42	* dictionary
43	* Handle the fallback if all the pages in blocklet are not
44	* encoded with local dictionary
45	*/
46	public class BlockletEncodedColumnPage {
47
48	/**
49	* LOGGER
50	*/
51	private static final Logger LOGGER =	1✔
52	LogServiceFactory.getLogService(BlockletEncodedColumnPage.class.getName());	1✔
53
54	/**
55	* list of encoded page of a column in a blocklet
56	*/
57	private List<EncodedColumnPage> encodedColumnPageList;
58
59	/**
60	* fallback executor service
61	*/
62	private ExecutorService fallbackExecutorService;
63
64	/**
65	* to check whether pages are local dictionary encoded or not
66	*/
67	private boolean isLocalDictEncoded;
68
69	/**
70	* page level dictionary only when column is encoded with local dictionary
71	*/
72	private PageLevelDictionary pageLevelDictionary;
73
74	/**
75	* fallback future task queue;
76	*/
77	private ArrayDeque<Future<FallbackEncodedColumnPage>> fallbackFutureQueue;
78
79	private String columnName;
80
81	/**
82	* is decoder based fallback enabled
83	*/
84	private boolean isDecoderBasedFallBackEnabled;
85
86	/**
87	* Local dictionary generator for column
88	*/
89	private LocalDictionaryGenerator localDictionaryGenerator;
90
91	BlockletEncodedColumnPage(ExecutorService fallbackExecutorService,
92	boolean isDecoderBasedFallBackEnabled, LocalDictionaryGenerator localDictionaryGenerator) {	1✔
93	this.fallbackExecutorService = fallbackExecutorService;	1✔
94	this.fallbackFutureQueue = new ArrayDeque<>();	1✔
95	this.isDecoderBasedFallBackEnabled = isDecoderBasedFallBackEnabled;	1✔
96	this.localDictionaryGenerator = localDictionaryGenerator;	1✔
97	}	1✔
98
99	/**
100	* Below method will be used to add column page of a column
101	*
102	* @param encodedColumnPage
103	* encoded column page
104	*/
105	void addEncodedColumnPage(EncodedColumnPage encodedColumnPage) {
106	if (null == encodedColumnPageList) {	1✔
107	this.encodedColumnPageList = new ArrayList<>();	1✔
108	// if dimension page is local dictionary enabled and encoded with local dictionary
109	if (encodedColumnPage.isLocalDictGeneratedPage()) {	1✔
110	this.isLocalDictEncoded = true;	1✔
111	// get first page dictionary
112	this.pageLevelDictionary = encodedColumnPage.getPageDictionary();	1✔
113	}
114	this.encodedColumnPageList.add(encodedColumnPage);	1✔
115	this.columnName = encodedColumnPage.getActualPage().getColumnSpec().getFieldName();	1✔
116	return;	1✔
117	}
118	// when first page was encoded without dictionary and next page encoded with dictionary
119	// in a blocklet
120	if (!isLocalDictEncoded && encodedColumnPage.isLocalDictGeneratedPage()) {	1✔
121	LOGGER.info(	×
122	"Local dictionary Fallback is initiated for column: " + this.columnName + " for page:"
123	+ encodedColumnPageList.size());	×
124	initiateFallBack(encodedColumnPage, encodedColumnPageList.size());	×
125	// fill null so once page is decoded again fill the re-encoded page again
126	this.encodedColumnPageList.add(null);	×
127	}
128	// if local dictionary is false or column is encoded with local dictionary then
129	// add a page
130	else if (!isLocalDictEncoded \|\| encodedColumnPage.isLocalDictGeneratedPage()) {	1✔
131	// merge page level dictionary values
132	if (null != this.pageLevelDictionary) {	1✔
133	pageLevelDictionary.mergerDictionaryValues(encodedColumnPage.getPageDictionary());	1✔
134	}
135	this.encodedColumnPageList.add(encodedColumnPage);	1✔
136	}
137	// if all the older pages were encoded with dictionary and new pages are without dictionary
138	else {
139	isLocalDictEncoded = false;	1✔
140	pageLevelDictionary = null;	1✔
141	LOGGER.info("Local dictionary Fallback is initiated for column: " + this.columnName	1✔
142	+ " for pages: 1 to " + encodedColumnPageList.size());	1✔
143	// submit all the older pages encoded with dictionary for fallback
144	for (int pageIndex = 0; pageIndex < encodedColumnPageList.size(); pageIndex++) {	1✔
145	if (encodedColumnPageList.get(pageIndex).getActualPage().isLocalDictGeneratedPage()) {	1✔
146	initiateFallBack(encodedColumnPageList.get(pageIndex), pageIndex);	1✔
147	}
148	}
149	//add to page list
150	this.encodedColumnPageList.add(encodedColumnPage);	1✔
151	}
152	}	1✔
153
154	/**
155	* Return the list of encoded page list for a column in a blocklet
156	*
157	* @return list of encoded page list
158	*/
159	public List<EncodedColumnPage> getEncodedColumnPageList() {
160	// if fallback queue is null then for some pages fallback was initiated
161	if (null != this.fallbackFutureQueue) {	1✔
162	try {
163	// check if queue is not empty
164	while (!fallbackFutureQueue.isEmpty()) {	1✔
165	// get the head element of queue
166	FallbackEncodedColumnPage fallbackEncodedColumnPage = fallbackFutureQueue.poll().get();	1✔
167	// add the encoded column page to list
168	encodedColumnPageList.set(fallbackEncodedColumnPage.getPageIndex(),	1✔
169	fallbackEncodedColumnPage.getEncodedColumnPage());	1✔
170	}	1✔
171	} catch (ExecutionException \| InterruptedException e) {	×
172	throw new RuntimeException("Problem while encoding the blocklet data during fallback", e);	×
173	}	1✔
174	// setting to null as all the fallback encoded page has been added to list
175	fallbackFutureQueue = null;	1✔
176	}
177	// in case of dictionary encoded column page memory will be freed only after
178	// all the pages are added in a blocklet, as fallback can happen anytime so old pages memory
179	// cannot be freed, so after encoding is done we can free the page memory
180	if (null != pageLevelDictionary) {	1✔
181	// clear the memory footprint for local dictionary encoded pages
182	for (EncodedColumnPage columnPage : encodedColumnPageList) {	1✔
183	columnPage.freeMemory();	1✔
184	}	1✔
185	}
186	return encodedColumnPageList;	1✔
187	}
188
189	/**
190	* Below method will be used to get the encoded dictionary
191	* values for local dictionary generated columns
192	*
193	* @return encoded dictionary values if column is local dictionary generated
194	*/
195	public LocalDictionaryChunk getEncodedDictionary() {
196	if (null != pageLevelDictionary) {	1✔
197	try {
198	return pageLevelDictionary.getLocalDictionaryChunkForBlocklet();	1✔
199	} catch (IOException \| MemoryException e) {	×
200	throw new RuntimeException(e);	×
201	}
202	}
203	return null;	1✔
204	}
205
206	/**
207	* This method initiates the fallback for local dictionary encoded column page
208	* @param encodedColumnPage
209	* @param pageIndex
210	*/
211	private void initiateFallBack(EncodedColumnPage encodedColumnPage, int pageIndex) {
212	if (isDecoderBasedFallBackEnabled) {	1✔
213	fallbackFutureQueue.add(fallbackExecutorService.submit(	1✔
214	new DecoderBasedFallbackEncoder(encodedColumnPage, pageIndex, localDictionaryGenerator)));
215	} else {
216	fallbackFutureQueue.add(fallbackExecutorService.submit(	×
217	new ActualDataBasedFallbackEncoder(encodedColumnPage, pageIndex)));
218	}
219	}	1✔
220	}

apache / carbondata / 2204

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