6747

Build # 6747

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/539a690e3677b764565df3687ea5b310feefedf3">539a690e3<a href="https://github.com/apache/carbondata/commit/539a690e3677b764565df3687ea5b310feefedf3">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/23abf035b583ba86c88da0a845d5e328778be2a2">23abf035b</a>">23abf035b</a><a href="https://github.com/apache/carbondata/commit/539a690e3677b764565df3687ea5b310feefedf3"> into 6029b2800">6029b2800</a>

Run Details

576 of 576 new or added lines in 25 files covered. (100.0%)

57160 of 70979 relevant lines covered (80.53%)

1.07 hits per line

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

0.0

/store/search/src/main/java/org/apache/carbondata/store/worker/SearchRequestHandler.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.store.worker;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;

import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.logging.LogService;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.datamap.DataMapChooser;
import org.apache.carbondata.core.datamap.DataMapDistributable;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.dev.expr.DataMapDistributableWrapper;
import org.apache.carbondata.core.datamap.dev.expr.DataMapExprWrapper;
import org.apache.carbondata.core.datastore.block.TableBlockInfo;
import org.apache.carbondata.core.datastore.row.CarbonRow;
import org.apache.carbondata.core.indexstore.ExtendedBlocklet;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.TableInfo;
import org.apache.carbondata.core.readcommitter.TableStatusReadCommittedScope;
import org.apache.carbondata.core.scan.executor.impl.SearchModeDetailQueryExecutor;
import org.apache.carbondata.core.scan.executor.impl.SearchModeVectorDetailQueryExecutor;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.scan.model.QueryModel;
import org.apache.carbondata.core.scan.model.QueryModelBuilder;
import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
import org.apache.carbondata.core.util.CarbonTaskInfo;
import org.apache.carbondata.core.util.ThreadLocalTaskInfo;
import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.hadoop.CarbonInputSplit;
import org.apache.carbondata.hadoop.CarbonMultiBlockSplit;
import org.apache.carbondata.hadoop.CarbonRecordReader;
import org.apache.carbondata.hadoop.readsupport.impl.CarbonRowReadSupport;

import org.apache.spark.search.SearchRequest;
import org.apache.spark.search.SearchResult;
import org.apache.spark.search.ShutdownRequest;
import org.apache.spark.search.ShutdownResponse;

/**
 * Thread runnable for handling SearchRequest from master.
 */
@InterfaceAudience.Internal
public class SearchRequestHandler {

  private static final LogService LOG =
      LogServiceFactory.getLogService(SearchRequestHandler.class.getName());

  public SearchResult handleSearch(SearchRequest request) {
    try {
      LOG.info(String.format("[SearchId:%d] receive search request", request.searchId()));
      List<CarbonRow> rows = handleRequest(request);
      LOG.info(String.format("[SearchId:%d] sending success response", request.searchId()));
      return createSuccessResponse(request, rows);
    } catch (IOException | InterruptedException e) {
      LOG.error(e);
      LOG.info(String.format("[SearchId:%d] sending failure response", request.searchId()));
      return createFailureResponse(request, e);
    }
  }

  public ShutdownResponse handleShutdown(ShutdownRequest request) {
    LOG.info("Shutting down worker...");
    SearchModeDetailQueryExecutor.shutdownThreadPool();
    SearchModeVectorDetailQueryExecutor.shutdownThreadPool();
    LOG.info("Worker shutted down");
    return new ShutdownResponse(Status.SUCCESS.ordinal(), "");
  }

  private DataMapExprWrapper chooseFGDataMap(
          CarbonTable table,
          FilterResolverIntf filterInterface) {
    DataMapChooser chooser = null;
    try {
      chooser = new DataMapChooser(table);
      return chooser.chooseFGDataMap(filterInterface);
    } catch (IOException e) {
      LOG.audit(e.getMessage());
      return null;
    }
  }

  /**
   * Builds {@link QueryModel} and read data from files
   */
  private List<CarbonRow> handleRequest(SearchRequest request)
      throws IOException, InterruptedException {
    CarbonTaskInfo carbonTaskInfo = new CarbonTaskInfo();
    carbonTaskInfo.setTaskId(System.nanoTime());
    ThreadLocalTaskInfo.setCarbonTaskInfo(carbonTaskInfo);
    TableInfo tableInfo = request.tableInfo();
    CarbonTable table = CarbonTable.buildFromTableInfo(tableInfo);
    QueryModel queryModel = createQueryModel(table, request);

    // in search mode, plain reader is better since it requires less memory
    queryModel.setVectorReader(false);

    CarbonMultiBlockSplit mbSplit = request.split().value();
    List<TableBlockInfo> list = CarbonInputSplit.createBlocks(mbSplit.getAllSplits());
    queryModel.setTableBlockInfos(list);
    long limit = request.limit();
    long rowCount = 0;

    LOG.info(String.format("[SearchId:%d] %s, number of block: %d",
        request.searchId(), queryModel.toString(), mbSplit.getAllSplits().size()));
    DataMapExprWrapper fgDataMap = chooseFGDataMap(table,
            queryModel.getFilterExpressionResolverTree());

    // If there is DataMap selected in Master, prune the split by it
    if (fgDataMap != null) {
      queryModel = prune(request.searchId(), table, queryModel, mbSplit, fgDataMap);
    }

    // In search mode, reader will read multiple blocks by using a thread pool
    CarbonRecordReader<CarbonRow> reader =
        new CarbonRecordReader<>(queryModel, new CarbonRowReadSupport());

    // read all rows by the reader
    List<CarbonRow> rows = new LinkedList<>();
    try {
      reader.initialize(mbSplit, null);

      // loop to read required number of rows.
      // By default, if user does not specify the limit value, limit is Long.MaxValue
      while (reader.nextKeyValue() && rowCount < limit) {
        rows.add(reader.getCurrentValue());
        rowCount++;
      }
    } catch (InterruptedException e) {
      throw new IOException(e);
    } finally {
      reader.close();
    }
    LOG.info(String.format("[SearchId:%d] scan completed, return %d rows",
        request.searchId(), rows.size()));
    return rows;
  }

  /**
   * If there is FGDataMap defined for this table and filter condition in the query,
   * prune the splits by the DataMap and set the pruned split into the QueryModel and return
   */
  private QueryModel prune(int queryId, CarbonTable table, QueryModel queryModel,
      CarbonMultiBlockSplit mbSplit, DataMapExprWrapper datamap) throws IOException {
    Objects.requireNonNull(datamap);
    List<Segment> segments = new LinkedList<>();
    HashMap<String, Integer> uniqueSegments = new HashMap<>();
    LoadMetadataDetails[] loadMetadataDetails =
        SegmentStatusManager.readLoadMetadata(
            CarbonTablePath.getMetadataPath(table.getTablePath()));
    for (CarbonInputSplit split : mbSplit.getAllSplits()) {
      String segmentId = Segment.getSegment(split.getSegmentId(), loadMetadataDetails).toString();
      if (uniqueSegments.get(segmentId) == null) {
        segments.add(Segment.toSegment(segmentId,
            new TableStatusReadCommittedScope(table.getAbsoluteTableIdentifier(),
                loadMetadataDetails)));
        uniqueSegments.put(segmentId, 1);
      } else {
        uniqueSegments.put(segmentId, uniqueSegments.get(segmentId) + 1);
      }
    }

    List<DataMapDistributableWrapper> distributables = datamap.toDistributable(segments);
    List<ExtendedBlocklet> prunnedBlocklets = new LinkedList<ExtendedBlocklet>();
    for (int i = 0; i < distributables.size(); i++) {
      DataMapDistributable dataMapDistributable = distributables.get(i).getDistributable();
      prunnedBlocklets.addAll(datamap.prune(dataMapDistributable, null));
    }

    HashMap<String, ExtendedBlocklet> pathToRead = new HashMap<>();
    for (ExtendedBlocklet prunedBlocklet : prunnedBlocklets) {
      pathToRead.put(prunedBlocklet.getFilePath().replace('\\', '/'), prunedBlocklet);
    }

    List<TableBlockInfo> blocks = queryModel.getTableBlockInfos();
    List<TableBlockInfo> blockToRead = new LinkedList<>();
    for (TableBlockInfo block : blocks) {
      if (pathToRead.keySet().contains(block.getFilePath())) {
        // If not set this, it will can't create FineGrainBlocklet object in
        // org.apache.carbondata.core.indexstore.blockletindex.BlockletDataRefNode.getIndexedData
        block.setDataMapWriterPath(pathToRead.get(block.getFilePath()).getDataMapWriterPath());
        blockToRead.add(block);
      }
    }
    LOG.info(String.format("[SearchId:%d] pruned using FG DataMap, pruned blocks: %d", queryId,
        blockToRead.size()));
    queryModel.setTableBlockInfos(blockToRead);
    queryModel.setFG(true);
    return queryModel;
  }

  private QueryModel createQueryModel(CarbonTable table, SearchRequest request) {
    String[] projectColumns = request.projectColumns();
    Expression filter = null;
    if (request.filterExpression() != null) {
      filter = request.filterExpression();
    }
    return new QueryModelBuilder(table)
        .projectColumns(projectColumns)
        .filterExpression(filter)
        .build();
  }

  /**
   * create a failure response
   */
  private SearchResult createFailureResponse(SearchRequest request, Throwable throwable) {
    return new SearchResult(request.searchId(), Status.FAILURE.ordinal(), throwable.getMessage(),
        new Object[0][]);
  }

  /**
   * create a success response with result rows
   */
  private SearchResult createSuccessResponse(SearchRequest request, List<CarbonRow> rows) {
    Iterator<CarbonRow> itor = rows.iterator();
    Object[][] output = new Object[rows.size()][];
    int i = 0;
    while (itor.hasNext()) {
      output[i++] = itor.next().getData();
    }
    return new SearchResult(request.searchId(), Status.SUCCESS.ordinal(), "", output);
  }

}

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.store.worker;
19
20	import java.io.IOException;
21	import java.util.HashMap;
22	import java.util.Iterator;
23	import java.util.LinkedList;
24	import java.util.List;
25	import java.util.Objects;
26
27	import org.apache.carbondata.common.annotations.InterfaceAudience;
28	import org.apache.carbondata.common.logging.LogService;
29	import org.apache.carbondata.common.logging.LogServiceFactory;
30	import org.apache.carbondata.core.datamap.DataMapChooser;
31	import org.apache.carbondata.core.datamap.DataMapDistributable;
32	import org.apache.carbondata.core.datamap.Segment;
33	import org.apache.carbondata.core.datamap.dev.expr.DataMapDistributableWrapper;
34	import org.apache.carbondata.core.datamap.dev.expr.DataMapExprWrapper;
35	import org.apache.carbondata.core.datastore.block.TableBlockInfo;
36	import org.apache.carbondata.core.datastore.row.CarbonRow;
37	import org.apache.carbondata.core.indexstore.ExtendedBlocklet;
38	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
39	import org.apache.carbondata.core.metadata.schema.table.TableInfo;
40	import org.apache.carbondata.core.readcommitter.TableStatusReadCommittedScope;
41	import org.apache.carbondata.core.scan.executor.impl.SearchModeDetailQueryExecutor;
42	import org.apache.carbondata.core.scan.executor.impl.SearchModeVectorDetailQueryExecutor;
43	import org.apache.carbondata.core.scan.expression.Expression;
44	import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
45	import org.apache.carbondata.core.scan.model.QueryModel;
46	import org.apache.carbondata.core.scan.model.QueryModelBuilder;
47	import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
48	import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
49	import org.apache.carbondata.core.util.CarbonTaskInfo;
50	import org.apache.carbondata.core.util.ThreadLocalTaskInfo;
51	import org.apache.carbondata.core.util.path.CarbonTablePath;
52	import org.apache.carbondata.hadoop.CarbonInputSplit;
53	import org.apache.carbondata.hadoop.CarbonMultiBlockSplit;
54	import org.apache.carbondata.hadoop.CarbonRecordReader;
55	import org.apache.carbondata.hadoop.readsupport.impl.CarbonRowReadSupport;
56
57	import org.apache.spark.search.SearchRequest;
58	import org.apache.spark.search.SearchResult;
59	import org.apache.spark.search.ShutdownRequest;
60	import org.apache.spark.search.ShutdownResponse;
61
62	/**
63	* Thread runnable for handling SearchRequest from master.
64	*/
65	@InterfaceAudience.Internal
66	public class SearchRequestHandler {	×
67
68	private static final LogService LOG =	×
69	LogServiceFactory.getLogService(SearchRequestHandler.class.getName());	×
70
71	public SearchResult handleSearch(SearchRequest request) {
72	try {
73	LOG.info(String.format("[SearchId:%d] receive search request", request.searchId()));	×
74	List<CarbonRow> rows = handleRequest(request);	×
75	LOG.info(String.format("[SearchId:%d] sending success response", request.searchId()));	×
76	return createSuccessResponse(request, rows);	×
77	} catch (IOException \| InterruptedException e) {	×
78	LOG.error(e);	×
79	LOG.info(String.format("[SearchId:%d] sending failure response", request.searchId()));	×
80	return createFailureResponse(request, e);	×
81	}
82	}
83
84	public ShutdownResponse handleShutdown(ShutdownRequest request) {
85	LOG.info("Shutting down worker...");	×
86	SearchModeDetailQueryExecutor.shutdownThreadPool();	×
87	SearchModeVectorDetailQueryExecutor.shutdownThreadPool();	×
88	LOG.info("Worker shutted down");	×
89	return new ShutdownResponse(Status.SUCCESS.ordinal(), "");	×
90	}
91
92	private DataMapExprWrapper chooseFGDataMap(
93	CarbonTable table,
94	FilterResolverIntf filterInterface) {
95	DataMapChooser chooser = null;	×
96	try {
97	chooser = new DataMapChooser(table);	×
98	return chooser.chooseFGDataMap(filterInterface);	×
99	} catch (IOException e) {	×
100	LOG.audit(e.getMessage());	×
101	return null;	×
102	}
103	}
104
105	/**
106	* Builds {@link QueryModel} and read data from files
107	*/
108	private List<CarbonRow> handleRequest(SearchRequest request)
109	throws IOException, InterruptedException {
110	CarbonTaskInfo carbonTaskInfo = new CarbonTaskInfo();	×
111	carbonTaskInfo.setTaskId(System.nanoTime());	×
112	ThreadLocalTaskInfo.setCarbonTaskInfo(carbonTaskInfo);	×
113	TableInfo tableInfo = request.tableInfo();	×
114	CarbonTable table = CarbonTable.buildFromTableInfo(tableInfo);	×
115	QueryModel queryModel = createQueryModel(table, request);	×
116
117	// in search mode, plain reader is better since it requires less memory
118	queryModel.setVectorReader(false);	×
119
120	CarbonMultiBlockSplit mbSplit = request.split().value();	×
121	List<TableBlockInfo> list = CarbonInputSplit.createBlocks(mbSplit.getAllSplits());	×
122	queryModel.setTableBlockInfos(list);	×
123	long limit = request.limit();	×
124	long rowCount = 0;	×
125
126	LOG.info(String.format("[SearchId:%d] %s, number of block: %d",	×
127	request.searchId(), queryModel.toString(), mbSplit.getAllSplits().size()));	×
128	DataMapExprWrapper fgDataMap = chooseFGDataMap(table,	×
129	queryModel.getFilterExpressionResolverTree());	×
130
131	// If there is DataMap selected in Master, prune the split by it
132	if (fgDataMap != null) {	×
133	queryModel = prune(request.searchId(), table, queryModel, mbSplit, fgDataMap);	×
134	}
135
136	// In search mode, reader will read multiple blocks by using a thread pool
137	CarbonRecordReader<CarbonRow> reader =	×
138	new CarbonRecordReader<>(queryModel, new CarbonRowReadSupport());
139
140	// read all rows by the reader
141	List<CarbonRow> rows = new LinkedList<>();	×
142	try {
143	reader.initialize(mbSplit, null);	×
144
145	// loop to read required number of rows.
146	// By default, if user does not specify the limit value, limit is Long.MaxValue
147	while (reader.nextKeyValue() && rowCount < limit) {	×
148	rows.add(reader.getCurrentValue());	×
149	rowCount++;	×
150	}
151	} catch (InterruptedException e) {	×
152	throw new IOException(e);	×
153	} finally {
154	reader.close();	×
155	}	×
156	LOG.info(String.format("[SearchId:%d] scan completed, return %d rows",	×
157	request.searchId(), rows.size()));	×
158	return rows;	×
159	}
160
161	/**
162	* If there is FGDataMap defined for this table and filter condition in the query,
163	* prune the splits by the DataMap and set the pruned split into the QueryModel and return
164	*/
165	private QueryModel prune(int queryId, CarbonTable table, QueryModel queryModel,
166	CarbonMultiBlockSplit mbSplit, DataMapExprWrapper datamap) throws IOException {
167	Objects.requireNonNull(datamap);	×
168	List<Segment> segments = new LinkedList<>();	×
169	HashMap<String, Integer> uniqueSegments = new HashMap<>();	×
170	LoadMetadataDetails[] loadMetadataDetails =	×
171	SegmentStatusManager.readLoadMetadata(	×
172	CarbonTablePath.getMetadataPath(table.getTablePath()));	×
173	for (CarbonInputSplit split : mbSplit.getAllSplits()) {	×
174	String segmentId = Segment.getSegment(split.getSegmentId(), loadMetadataDetails).toString();	×
175	if (uniqueSegments.get(segmentId) == null) {	×
176	segments.add(Segment.toSegment(segmentId,	×
177	new TableStatusReadCommittedScope(table.getAbsoluteTableIdentifier(),	×
178	loadMetadataDetails)));
179	uniqueSegments.put(segmentId, 1);	×
180	} else {
181	uniqueSegments.put(segmentId, uniqueSegments.get(segmentId) + 1);	×
182	}
183	}	×
184
185	List<DataMapDistributableWrapper> distributables = datamap.toDistributable(segments);	×
186	List<ExtendedBlocklet> prunnedBlocklets = new LinkedList<ExtendedBlocklet>();	×
187	for (int i = 0; i < distributables.size(); i++) {	×
188	DataMapDistributable dataMapDistributable = distributables.get(i).getDistributable();	×
189	prunnedBlocklets.addAll(datamap.prune(dataMapDistributable, null));	×
190	}
191
192	HashMap<String, ExtendedBlocklet> pathToRead = new HashMap<>();	×
193	for (ExtendedBlocklet prunedBlocklet : prunnedBlocklets) {	×
194	pathToRead.put(prunedBlocklet.getFilePath().replace('\\', '/'), prunedBlocklet);	×
195	}	×
196
197	List<TableBlockInfo> blocks = queryModel.getTableBlockInfos();	×
198	List<TableBlockInfo> blockToRead = new LinkedList<>();	×
199	for (TableBlockInfo block : blocks) {	×
200	if (pathToRead.keySet().contains(block.getFilePath())) {	×
201	// If not set this, it will can't create FineGrainBlocklet object in
202	// org.apache.carbondata.core.indexstore.blockletindex.BlockletDataRefNode.getIndexedData
203	block.setDataMapWriterPath(pathToRead.get(block.getFilePath()).getDataMapWriterPath());	×
204	blockToRead.add(block);	×
205	}
206	}	×
207	LOG.info(String.format("[SearchId:%d] pruned using FG DataMap, pruned blocks: %d", queryId,	×
208	blockToRead.size()));	×
209	queryModel.setTableBlockInfos(blockToRead);	×
210	queryModel.setFG(true);	×
211	return queryModel;	×
212	}
213
214	private QueryModel createQueryModel(CarbonTable table, SearchRequest request) {
215	String[] projectColumns = request.projectColumns();	×
216	Expression filter = null;	×
217	if (request.filterExpression() != null) {	×
218	filter = request.filterExpression();	×
219	}
220	return new QueryModelBuilder(table)	×
221	.projectColumns(projectColumns)	×
222	.filterExpression(filter)	×
223	.build();	×
224	}
225
226	/**
227	* create a failure response
228	*/
229	private SearchResult createFailureResponse(SearchRequest request, Throwable throwable) {
230	return new SearchResult(request.searchId(), Status.FAILURE.ordinal(), throwable.getMessage(),	×
231	new Object[0][]);
232	}
233
234	/**
235	* create a success response with result rows
236	*/
237	private SearchResult createSuccessResponse(SearchRequest request, List<CarbonRow> rows) {
238	Iterator<CarbonRow> itor = rows.iterator();	×
239	Object[][] output = new Object[rows.size()][];	×
240	int i = 0;	×
241	while (itor.hasNext()) {	×
242	output[i++] = itor.next().getData();	×
243	}
244	return new SearchResult(request.searchId(), Status.SUCCESS.ordinal(), "", output);	×
245	}
246
247	}

apache / carbondata / 6747

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