11616

Build # 11616

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/97ee90d35e1d3c2f6fb2f43b86ea7df14b030192">97ee90d35">&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/97ee90d35e1d3c2f6fb2f43b86ea7df14b030192">97ee90d35</a><a href="https://github.com/apache/carbondata/commit/97ee90d35e1d3c2f6fb2f43b86ea7df14b030192">&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/97ee90d35e1d3c2f6fb2f43b86ea7df14b030192&quot;&gt;97ee90d35&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/97ee90d35e1d3c2f6fb2f43b86ea7df14b030192&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

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

64139 of 82063 relevant lines covered (78.16%)

1.03 hits per line

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

73.79

/core/src/main/java/org/apache/carbondata/core/datamap/DataMapUtil.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.datamap;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.datamap.dev.DataMap;
import org.apache.carbondata.core.datamap.dev.expr.DataMapDistributableWrapper;
import org.apache.carbondata.core.datamap.dev.expr.DataMapExprWrapper;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.indexstore.ExtendedBlocklet;
import org.apache.carbondata.core.indexstore.PartitionSpec;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.RelationIdentifier;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
import org.apache.carbondata.core.util.ObjectSerializationUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;

import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;

public class DataMapUtil {

  private static final String DATA_MAP_DSTR = "mapreduce.input.carboninputformat.datamapdstr";

  public static final String EMBEDDED_JOB_NAME =
      "org.apache.carbondata.indexserver.EmbeddedDataMapJob";

  public static final String DISTRIBUTED_JOB_NAME =
      "org.apache.carbondata.indexserver.DistributedDataMapJob";

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(DataMapUtil.class.getName());

  /**
   * Creates instance for the DataMap Job class
   *
   * @param className
   * @return
   */
  public static Object createDataMapJob(String className) {
    try {
      return Class.forName(className).getDeclaredConstructors()[0].newInstance();
    } catch (Exception e) {
      LOGGER.error(e.getMessage(), e);
      return null;
    }
  }

  /**
   * This method sets the datamapJob in the configuration
   * @param configuration
   * @param dataMapJob
   * @throws IOException
   */
  public static void setDataMapJob(Configuration configuration, Object dataMapJob)
      throws IOException {
    if (dataMapJob != null) {
      String toString = ObjectSerializationUtil.convertObjectToString(dataMapJob);
      configuration.set(DATA_MAP_DSTR, toString);
    }
  }

  /**
   * get datamap job from the configuration
   * @param configuration job configuration
   * @return DataMap Job
   * @throws IOException
   */
  public static DataMapJob getDataMapJob(Configuration configuration) throws IOException {
    String jobString = configuration.get(DATA_MAP_DSTR);
    if (jobString != null) {
      return (DataMapJob) ObjectSerializationUtil.convertStringToObject(jobString);
    }
    return null;
  }

  /**
   * This method gets the datamapJob and call execute , this job will be launched before clearing
   * datamaps from driver side during drop table and drop datamap and clears the datamap in executor
   * side
   * @param carbonTable
   * @throws IOException
   */
  private static void executeClearDataMapJob(DataMapJob dataMapJob,
      CarbonTable carbonTable, String dataMapToClear) throws IOException {
    SegmentStatusManager.ValidAndInvalidSegmentsInfo validAndInvalidSegmentsInfo =
            getValidAndInvalidSegments(carbonTable, FileFactory.getConfiguration());
    List<String> invalidSegment = new ArrayList<>();
    for (Segment segment : validAndInvalidSegmentsInfo.getInvalidSegments()) {
      invalidSegment.add(segment.getSegmentNo());
    }
    DistributableDataMapFormat dataMapFormat = new DistributableDataMapFormat(carbonTable,
        validAndInvalidSegmentsInfo.getValidSegments(), invalidSegment, true,
        dataMapToClear);
    try {
      dataMapJob.execute(dataMapFormat);
    } catch (Exception e) {
      if (dataMapJob.getClass().getName().equalsIgnoreCase(DISTRIBUTED_JOB_NAME)) {
        LOGGER.warn("Failed to clear distributed cache.", e);
      } else {
        throw e;
      }
    }
  }

  public static void executeClearDataMapJob(CarbonTable carbonTable, String jobClassName)
      throws IOException {
    executeClearDataMapJob(carbonTable, jobClassName, "");
  }

  static void executeClearDataMapJob(CarbonTable carbonTable, String jobClassName,
      String dataMapToClear) throws IOException {
    DataMapJob dataMapJob = (DataMapJob) createDataMapJob(jobClassName);
    if (dataMapJob == null) {
      return;
    }
    executeClearDataMapJob(dataMapJob, carbonTable, dataMapToClear);
  }

  public static DataMapJob getEmbeddedJob() {
    DataMapJob dataMapJob = (DataMapJob) DataMapUtil.createDataMapJob(EMBEDDED_JOB_NAME);
    if (dataMapJob == null) {
      throw new ExceptionInInitializerError("Unable to create EmbeddedDataMapJob");
    }
    return dataMapJob;
  }

  /**
   * Prune the segments from the already pruned blocklets.
   */
  public static void pruneSegments(List<Segment> segments, List<ExtendedBlocklet> prunedBlocklets) {
    Set<Segment> validSegments = new HashSet<>();
    for (ExtendedBlocklet blocklet : prunedBlocklets) {
      // Clear the old pruned index files if any present
      blocklet.getSegment().getFilteredIndexShardNames().clear();
      // Set the pruned index file to the segment
      // for further pruning.
      String shardName = CarbonTablePath.getShardName(blocklet.getFilePath());
      blocklet.getSegment().setFilteredIndexShardName(shardName);
      validSegments.add(blocklet.getSegment());
    }
    segments.clear();
    segments.addAll(validSegments);
  }

  static List<ExtendedBlocklet> pruneDataMaps(CarbonTable table,
      FilterResolverIntf filterResolverIntf, List<Segment> segmentsToLoad,
      List<PartitionSpec> partitions, List<ExtendedBlocklet> blocklets) throws IOException {
    pruneSegments(segmentsToLoad, blocklets);
    List<ExtendedBlocklet> cgDataMaps = pruneDataMaps(table, filterResolverIntf, segmentsToLoad,
        partitions, blocklets,
        DataMapLevel.CG);
    pruneSegments(segmentsToLoad, cgDataMaps);
    return pruneDataMaps(table, filterResolverIntf, segmentsToLoad,
        partitions, cgDataMaps,
        DataMapLevel.FG);
  }

  static List<ExtendedBlocklet> pruneDataMaps(CarbonTable table,
      FilterResolverIntf filterResolverIntf, List<Segment> segmentsToLoad,
      List<PartitionSpec> partitions, List<ExtendedBlocklet> blocklets, DataMapLevel dataMapLevel)
      throws IOException {
    DataMapExprWrapper dataMapExprWrapper =
        new DataMapChooser(table).chooseDataMap(dataMapLevel, filterResolverIntf);
    if (dataMapExprWrapper != null) {
      List<ExtendedBlocklet> extendedBlocklets = new ArrayList<>();
      // Prune segments from already pruned blocklets
      for (DataMapDistributableWrapper wrapper : dataMapExprWrapper
          .toDistributable(segmentsToLoad)) {
        TableDataMap dataMap = DataMapStoreManager.getInstance()
            .getDataMap(table, wrapper.getDistributable().getDataMapSchema());
        List<DataMap> dataMaps = dataMap.getTableDataMaps(wrapper.getDistributable());
        List<ExtendedBlocklet> prunnedBlocklet = new ArrayList<>();
        if (table.isTransactionalTable()) {
          prunnedBlocklet.addAll(dataMap.prune(dataMaps, wrapper.getDistributable(),
              dataMapExprWrapper.getFilterResolverIntf(wrapper.getUniqueId()), partitions));
        } else {
          prunnedBlocklet
              .addAll(dataMap.prune(segmentsToLoad, new DataMapFilter(filterResolverIntf),
                  partitions));
        }
        // For all blocklets initialize the detail info so that it can be serialized to the driver.
        for (ExtendedBlocklet blocklet : prunnedBlocklet) {
          blocklet.getDetailInfo();
          blocklet.setDataMapUniqueId(wrapper.getUniqueId());
        }
        extendedBlocklets.addAll(prunnedBlocklet);
      }
      return dataMapExprWrapper.pruneBlocklets(extendedBlocklets);
    }
    // For all blocklets initialize the detail info so that it can be serialized to the driver.
    for (ExtendedBlocklet blocklet : blocklets) {
      blocklet.getDetailInfo();
    }
    return blocklets;
  }

  /**
   * this method gets the datamapJob and call execute of that job, this will be launched for
   * distributed CG or FG
   * @return list of Extended blocklets after pruning
   */
  public static List<ExtendedBlocklet> executeDataMapJob(CarbonTable carbonTable,
      FilterResolverIntf resolver, DataMapJob dataMapJob, List<PartitionSpec> partitionsToPrune,
      List<Segment> validSegments, List<Segment> invalidSegments, DataMapLevel level,
      List<String> segmentsToBeRefreshed) throws IOException {
    return executeDataMapJob(carbonTable, resolver, dataMapJob, partitionsToPrune, validSegments,
        invalidSegments, level, false, segmentsToBeRefreshed);
  }

  /**
   * this method gets the datamapJob and call execute of that job, this will be launched for
   * distributed CG or FG
   * @return list of Extended blocklets after pruning
   */
  public static List<ExtendedBlocklet> executeDataMapJob(CarbonTable carbonTable,
      FilterResolverIntf resolver, DataMapJob dataMapJob, List<PartitionSpec> partitionsToPrune,
      List<Segment> validSegments, List<Segment> invalidSegments, DataMapLevel level,
      Boolean isFallbackJob, List<String> segmentsToBeRefreshed) throws IOException {
    List<String> invalidSegmentNo = new ArrayList<>();
    for (Segment segment : invalidSegments) {
      invalidSegmentNo.add(segment.getSegmentNo());
    }
    invalidSegmentNo.addAll(segmentsToBeRefreshed);
    DistributableDataMapFormat dataMapFormat =
        new DistributableDataMapFormat(carbonTable, resolver, validSegments, invalidSegmentNo,
            partitionsToPrune, false, level, isFallbackJob);
    List<ExtendedBlocklet> prunedBlocklets = dataMapJob.execute(dataMapFormat);
    // Apply expression on the blocklets.
    return prunedBlocklets;
  }

  public static SegmentStatusManager.ValidAndInvalidSegmentsInfo getValidAndInvalidSegments(
      CarbonTable carbonTable, Configuration configuration) throws IOException {
    SegmentStatusManager ssm =
        new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier(), configuration);
    return ssm.getValidAndInvalidSegments();
  }

  /**
   * Returns valid segment list for a given RelationIdentifier
   *
   * @param relationIdentifier get list of segments for relation identifier
   * @return list of valid segment id's
   * @throws IOException
   */
  public static List<String> getMainTableValidSegmentList(RelationIdentifier relationIdentifier)
      throws IOException {
    List<String> segmentList = new ArrayList<>();
    List<Segment> validSegments = new SegmentStatusManager(AbsoluteTableIdentifier
        .from(relationIdentifier.getTablePath(), relationIdentifier.getDatabaseName(),
            relationIdentifier.getTableName())).getValidAndInvalidSegments().getValidSegments();
    for (Segment segment : validSegments) {
      segmentList.add(segment.getSegmentNo());
    }
    return segmentList;
  }

}

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.datamap;
19
20	import java.io.IOException;
21	import java.util.ArrayList;
22	import java.util.HashSet;
23	import java.util.List;
24	import java.util.Set;
25
26	import org.apache.carbondata.common.logging.LogServiceFactory;
27	import org.apache.carbondata.core.datamap.dev.DataMap;
28	import org.apache.carbondata.core.datamap.dev.expr.DataMapDistributableWrapper;
29	import org.apache.carbondata.core.datamap.dev.expr.DataMapExprWrapper;
30	import org.apache.carbondata.core.datastore.impl.FileFactory;
31	import org.apache.carbondata.core.indexstore.ExtendedBlocklet;
32	import org.apache.carbondata.core.indexstore.PartitionSpec;
33	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
34	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
35	import org.apache.carbondata.core.metadata.schema.table.RelationIdentifier;
36	import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
37	import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
38	import org.apache.carbondata.core.util.ObjectSerializationUtil;
39	import org.apache.carbondata.core.util.path.CarbonTablePath;
40
41	import org.apache.hadoop.conf.Configuration;
42	import org.apache.log4j.Logger;
43
44	public class DataMapUtil {	×
45
46	private static final String DATA_MAP_DSTR = "mapreduce.input.carboninputformat.datamapdstr";
47
48	public static final String EMBEDDED_JOB_NAME =
49	"org.apache.carbondata.indexserver.EmbeddedDataMapJob";
50
51	public static final String DISTRIBUTED_JOB_NAME =
52	"org.apache.carbondata.indexserver.DistributedDataMapJob";
53
54	private static final Logger LOGGER =	1✔
55	LogServiceFactory.getLogService(DataMapUtil.class.getName());	1✔
56
57	/**
58	* Creates instance for the DataMap Job class
59	*
60	* @param className
61	* @return
62	*/
63	public static Object createDataMapJob(String className) {
64	try {
65	return Class.forName(className).getDeclaredConstructors()[0].newInstance();	1✔
66	} catch (Exception e) {	1✔
67	LOGGER.error(e.getMessage(), e);	1✔
68	return null;	1✔
69	}
70	}
71
72	/**
73	* This method sets the datamapJob in the configuration
74	* @param configuration
75	* @param dataMapJob
76	* @throws IOException
77	*/
78	public static void setDataMapJob(Configuration configuration, Object dataMapJob)
79	throws IOException {
80	if (dataMapJob != null) {	1✔
81	String toString = ObjectSerializationUtil.convertObjectToString(dataMapJob);	1✔
82	configuration.set(DATA_MAP_DSTR, toString);	1✔
83	}
84	}	1✔
85
86	/**
87	* get datamap job from the configuration
88	* @param configuration job configuration
89	* @return DataMap Job
90	* @throws IOException
91	*/
92	public static DataMapJob getDataMapJob(Configuration configuration) throws IOException {
93	String jobString = configuration.get(DATA_MAP_DSTR);	1✔
94	if (jobString != null) {	1✔
95	return (DataMapJob) ObjectSerializationUtil.convertStringToObject(jobString);	1✔
96	}
97	return null;	1✔
98	}
99
100	/**
101	* This method gets the datamapJob and call execute , this job will be launched before clearing
102	* datamaps from driver side during drop table and drop datamap and clears the datamap in executor
103	* side
104	* @param carbonTable
105	* @throws IOException
106	*/
107	private static void executeClearDataMapJob(DataMapJob dataMapJob,
108	CarbonTable carbonTable, String dataMapToClear) throws IOException {
109	SegmentStatusManager.ValidAndInvalidSegmentsInfo validAndInvalidSegmentsInfo =	1✔
110	getValidAndInvalidSegments(carbonTable, FileFactory.getConfiguration());	1✔
111	List<String> invalidSegment = new ArrayList<>();	1✔
112	for (Segment segment : validAndInvalidSegmentsInfo.getInvalidSegments()) {	1✔
113	invalidSegment.add(segment.getSegmentNo());	1✔
114	}	1✔
115	DistributableDataMapFormat dataMapFormat = new DistributableDataMapFormat(carbonTable,	1✔
116	validAndInvalidSegmentsInfo.getValidSegments(), invalidSegment, true,	1✔
117	dataMapToClear);
118	try {
119	dataMapJob.execute(dataMapFormat);	1✔
120	} catch (Exception e) {	×
121	if (dataMapJob.getClass().getName().equalsIgnoreCase(DISTRIBUTED_JOB_NAME)) {	×
122	LOGGER.warn("Failed to clear distributed cache.", e);	×
123	} else {
124	throw e;	×
125	}
126	}	1✔
127	}	1✔
128
129	public static void executeClearDataMapJob(CarbonTable carbonTable, String jobClassName)
130	throws IOException {
131	executeClearDataMapJob(carbonTable, jobClassName, "");	1✔
132	}	1✔
133
134	static void executeClearDataMapJob(CarbonTable carbonTable, String jobClassName,
135	String dataMapToClear) throws IOException {
136	DataMapJob dataMapJob = (DataMapJob) createDataMapJob(jobClassName);	1✔
137	if (dataMapJob == null) {	1✔
138	return;	1✔
139	}
140	executeClearDataMapJob(dataMapJob, carbonTable, dataMapToClear);	1✔
141	}	1✔
142
143	public static DataMapJob getEmbeddedJob() {
144	DataMapJob dataMapJob = (DataMapJob) DataMapUtil.createDataMapJob(EMBEDDED_JOB_NAME);	×
145	if (dataMapJob == null) {	×
146	throw new ExceptionInInitializerError("Unable to create EmbeddedDataMapJob");	×
147	}
148	return dataMapJob;	×
149	}
150
151	/**
152	* Prune the segments from the already pruned blocklets.
153	*/
154	public static void pruneSegments(List<Segment> segments, List<ExtendedBlocklet> prunedBlocklets) {
155	Set<Segment> validSegments = new HashSet<>();	1✔
156	for (ExtendedBlocklet blocklet : prunedBlocklets) {	1✔
157	// Clear the old pruned index files if any present
158	blocklet.getSegment().getFilteredIndexShardNames().clear();	1✔
159	// Set the pruned index file to the segment
160	// for further pruning.
161	String shardName = CarbonTablePath.getShardName(blocklet.getFilePath());	1✔
162	blocklet.getSegment().setFilteredIndexShardName(shardName);	1✔
163	validSegments.add(blocklet.getSegment());	1✔
164	}	1✔
165	segments.clear();	1✔
166	segments.addAll(validSegments);	1✔
167	}	1✔
168
169	static List<ExtendedBlocklet> pruneDataMaps(CarbonTable table,
170	FilterResolverIntf filterResolverIntf, List<Segment> segmentsToLoad,
171	List<PartitionSpec> partitions, List<ExtendedBlocklet> blocklets) throws IOException {
172	pruneSegments(segmentsToLoad, blocklets);	×
173	List<ExtendedBlocklet> cgDataMaps = pruneDataMaps(table, filterResolverIntf, segmentsToLoad,	×
174	partitions, blocklets,
175	DataMapLevel.CG);
176	pruneSegments(segmentsToLoad, cgDataMaps);	×
177	return pruneDataMaps(table, filterResolverIntf, segmentsToLoad,	×
178	partitions, cgDataMaps,
179	DataMapLevel.FG);
180	}
181
182	static List<ExtendedBlocklet> pruneDataMaps(CarbonTable table,
183	FilterResolverIntf filterResolverIntf, List<Segment> segmentsToLoad,
184	List<PartitionSpec> partitions, List<ExtendedBlocklet> blocklets, DataMapLevel dataMapLevel)
185	throws IOException {
186	DataMapExprWrapper dataMapExprWrapper =	1✔
187	new DataMapChooser(table).chooseDataMap(dataMapLevel, filterResolverIntf);	1✔
188	if (dataMapExprWrapper != null) {	1✔
189	List<ExtendedBlocklet> extendedBlocklets = new ArrayList<>();	1✔
190	// Prune segments from already pruned blocklets
191	for (DataMapDistributableWrapper wrapper : dataMapExprWrapper	1✔
192	.toDistributable(segmentsToLoad)) {	1✔
193	TableDataMap dataMap = DataMapStoreManager.getInstance()	1✔
194	.getDataMap(table, wrapper.getDistributable().getDataMapSchema());	1✔
195	List<DataMap> dataMaps = dataMap.getTableDataMaps(wrapper.getDistributable());	1✔
196	List<ExtendedBlocklet> prunnedBlocklet = new ArrayList<>();	1✔
197	if (table.isTransactionalTable()) {	1✔
198	prunnedBlocklet.addAll(dataMap.prune(dataMaps, wrapper.getDistributable(),	1✔
199	dataMapExprWrapper.getFilterResolverIntf(wrapper.getUniqueId()), partitions));	1✔
200	} else {
201	prunnedBlocklet	×
202	.addAll(dataMap.prune(segmentsToLoad, new DataMapFilter(filterResolverIntf),	×
203	partitions));
204	}
205	// For all blocklets initialize the detail info so that it can be serialized to the driver.
206	for (ExtendedBlocklet blocklet : prunnedBlocklet) {	1✔
207	blocklet.getDetailInfo();	1✔
208	blocklet.setDataMapUniqueId(wrapper.getUniqueId());	1✔
209	}	1✔
210	extendedBlocklets.addAll(prunnedBlocklet);	1✔
211	}	1✔
212	return dataMapExprWrapper.pruneBlocklets(extendedBlocklets);	1✔
213	}
214	// For all blocklets initialize the detail info so that it can be serialized to the driver.
215	for (ExtendedBlocklet blocklet : blocklets) {	×
216	blocklet.getDetailInfo();	×
217	}	×
218	return blocklets;	×
219	}
220
221	/**
222	* this method gets the datamapJob and call execute of that job, this will be launched for
223	* distributed CG or FG
224	* @return list of Extended blocklets after pruning
225	*/
226	public static List<ExtendedBlocklet> executeDataMapJob(CarbonTable carbonTable,
227	FilterResolverIntf resolver, DataMapJob dataMapJob, List<PartitionSpec> partitionsToPrune,
228	List<Segment> validSegments, List<Segment> invalidSegments, DataMapLevel level,
229	List<String> segmentsToBeRefreshed) throws IOException {
230	return executeDataMapJob(carbonTable, resolver, dataMapJob, partitionsToPrune, validSegments,	1✔
231	invalidSegments, level, false, segmentsToBeRefreshed);	1✔
232	}
233
234	/**
235	* this method gets the datamapJob and call execute of that job, this will be launched for
236	* distributed CG or FG
237	* @return list of Extended blocklets after pruning
238	*/
239	public static List<ExtendedBlocklet> executeDataMapJob(CarbonTable carbonTable,
240	FilterResolverIntf resolver, DataMapJob dataMapJob, List<PartitionSpec> partitionsToPrune,
241	List<Segment> validSegments, List<Segment> invalidSegments, DataMapLevel level,
242	Boolean isFallbackJob, List<String> segmentsToBeRefreshed) throws IOException {
243	List<String> invalidSegmentNo = new ArrayList<>();	1✔
244	for (Segment segment : invalidSegments) {	1✔
245	invalidSegmentNo.add(segment.getSegmentNo());	1✔
246	}	1✔
247	invalidSegmentNo.addAll(segmentsToBeRefreshed);	1✔
248	DistributableDataMapFormat dataMapFormat =	1✔
249	new DistributableDataMapFormat(carbonTable, resolver, validSegments, invalidSegmentNo,
250	partitionsToPrune, false, level, isFallbackJob);	1✔
251	List<ExtendedBlocklet> prunedBlocklets = dataMapJob.execute(dataMapFormat);	1✔
252	// Apply expression on the blocklets.
253	return prunedBlocklets;	1✔
254	}
255
256	public static SegmentStatusManager.ValidAndInvalidSegmentsInfo getValidAndInvalidSegments(
257	CarbonTable carbonTable, Configuration configuration) throws IOException {
258	SegmentStatusManager ssm =	1✔
259	new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier(), configuration);	1✔
260	return ssm.getValidAndInvalidSegments();	1✔
261	}
262
263	/**
264	* Returns valid segment list for a given RelationIdentifier
265	*
266	* @param relationIdentifier get list of segments for relation identifier
267	* @return list of valid segment id's
268	* @throws IOException
269	*/
270	public static List<String> getMainTableValidSegmentList(RelationIdentifier relationIdentifier)
271	throws IOException {
272	List<String> segmentList = new ArrayList<>();	×
273	List<Segment> validSegments = new SegmentStatusManager(AbsoluteTableIdentifier	×
274	.from(relationIdentifier.getTablePath(), relationIdentifier.getDatabaseName(),	×
275	relationIdentifier.getTableName())).getValidAndInvalidSegments().getValidSegments();	×
276	for (Segment segment : validSegments) {	×
277	segmentList.add(segment.getSegmentNo());	×
278	}	×
279	return segmentList;	×
280	}
281
282	}

apache / carbondata / 11616

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