11940

Build # 11940

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/ea9600bc1ca19ef8d75a7c2c624e268b6c5d824b">ea9600bc1<a href="https://github.com/apache/carbondata/commit/ea9600bc1ca19ef8d75a7c2c624e268b6c5d824b">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/8c6ff810a975442b282a19cd7695e610e6ea6f80">8c6ff810a</a>">8c6ff810a</a><a href="https://github.com/apache/carbondata/commit/ea9600bc1ca19ef8d75a7c2c624e268b6c5d824b"> into 5a9d1e2e6">5a9d1e2e6</a>

Run Details

281 of 281 new or added lines in 12 files covered. (100.0%)

64844 of 83370 relevant lines covered (77.78%)

1.02 hits per line

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

74.51

/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,
      DataMapChooser dataMapChooser) throws IOException {
    if (null == dataMapChooser) {
      return blocklets;
    }
    pruneSegments(segmentsToLoad, blocklets);
    List<ExtendedBlocklet> cgDataMaps = pruneDataMaps(table, filterResolverIntf, segmentsToLoad,
        partitions, blocklets,
        DataMapLevel.CG, dataMapChooser);
    pruneSegments(segmentsToLoad, cgDataMaps);
    return pruneDataMaps(table, filterResolverIntf, segmentsToLoad,
        partitions, cgDataMaps,
        DataMapLevel.FG, dataMapChooser);
  }

  static List<ExtendedBlocklet> pruneDataMaps(CarbonTable table,
      FilterResolverIntf filterResolverIntf, List<Segment> segmentsToLoad,
      List<PartitionSpec> partitions, List<ExtendedBlocklet> blocklets, DataMapLevel dataMapLevel,
      DataMapChooser dataMapChooser)
      throws IOException {
    DataMapExprWrapper dataMapExprWrapper =
        dataMapChooser.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);
    }
    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,
172	DataMapChooser dataMapChooser) throws IOException {
173	if (null == dataMapChooser) {	×
174	return blocklets;	×
175	}
176	pruneSegments(segmentsToLoad, blocklets);	×
177	List<ExtendedBlocklet> cgDataMaps = pruneDataMaps(table, filterResolverIntf, segmentsToLoad,	×
178	partitions, blocklets,
179	DataMapLevel.CG, dataMapChooser);
180	pruneSegments(segmentsToLoad, cgDataMaps);	×
181	return pruneDataMaps(table, filterResolverIntf, segmentsToLoad,	×
182	partitions, cgDataMaps,
183	DataMapLevel.FG, dataMapChooser);
184	}
185
186	static List<ExtendedBlocklet> pruneDataMaps(CarbonTable table,
187	FilterResolverIntf filterResolverIntf, List<Segment> segmentsToLoad,
188	List<PartitionSpec> partitions, List<ExtendedBlocklet> blocklets, DataMapLevel dataMapLevel,
189	DataMapChooser dataMapChooser)
190	throws IOException {
191	DataMapExprWrapper dataMapExprWrapper =	1✔
192	dataMapChooser.chooseDataMap(dataMapLevel, filterResolverIntf);	1✔
193	if (dataMapExprWrapper != null) {	1✔
194	List<ExtendedBlocklet> extendedBlocklets = new ArrayList<>();	1✔
195	// Prune segments from already pruned blocklets
196	for (DataMapDistributableWrapper wrapper : dataMapExprWrapper	1✔
197	.toDistributable(segmentsToLoad)) {	1✔
198	TableDataMap dataMap = DataMapStoreManager.getInstance()	1✔
199	.getDataMap(table, wrapper.getDistributable().getDataMapSchema());	1✔
200	List<DataMap> dataMaps = dataMap.getTableDataMaps(wrapper.getDistributable());	1✔
201	List<ExtendedBlocklet> prunnedBlocklet = new ArrayList<>();	1✔
202	if (table.isTransactionalTable()) {	1✔
203	prunnedBlocklet.addAll(dataMap.prune(dataMaps, wrapper.getDistributable(),	1✔
204	dataMapExprWrapper.getFilterResolverIntf(wrapper.getUniqueId()), partitions));	1✔
205	} else {
206	prunnedBlocklet	×
207	.addAll(dataMap.prune(segmentsToLoad, new DataMapFilter(filterResolverIntf),	×
208	partitions));
209	}
210	// For all blocklets initialize the detail info so that it can be serialized to the driver.
211	for (ExtendedBlocklet blocklet : prunnedBlocklet) {	1✔
212	blocklet.getDetailInfo();	1✔
213	blocklet.setDataMapUniqueId(wrapper.getUniqueId());	1✔
214	}	1✔
215	extendedBlocklets.addAll(prunnedBlocklet);	1✔
216	}	1✔
217	return dataMapExprWrapper.pruneBlocklets(extendedBlocklets);	1✔
218	}
219	return blocklets;	×
220	}
221
222	/**
223	* this method gets the datamapJob and call execute of that job, this will be launched for
224	* distributed CG or FG
225	* @return list of Extended blocklets after pruning
226	*/
227	public static List<ExtendedBlocklet> executeDataMapJob(CarbonTable carbonTable,
228	FilterResolverIntf resolver, DataMapJob dataMapJob, List<PartitionSpec> partitionsToPrune,
229	List<Segment> validSegments, List<Segment> invalidSegments, DataMapLevel level,
230	List<String> segmentsToBeRefreshed) throws IOException {
231	return executeDataMapJob(carbonTable, resolver, dataMapJob, partitionsToPrune, validSegments,	1✔
232	invalidSegments, level, false, segmentsToBeRefreshed);	1✔
233	}
234
235	/**
236	* this method gets the datamapJob and call execute of that job, this will be launched for
237	* distributed CG or FG
238	* @return list of Extended blocklets after pruning
239	*/
240	public static List<ExtendedBlocklet> executeDataMapJob(CarbonTable carbonTable,
241	FilterResolverIntf resolver, DataMapJob dataMapJob, List<PartitionSpec> partitionsToPrune,
242	List<Segment> validSegments, List<Segment> invalidSegments, DataMapLevel level,
243	Boolean isFallbackJob, List<String> segmentsToBeRefreshed) throws IOException {
244	List<String> invalidSegmentNo = new ArrayList<>();	1✔
245	for (Segment segment : invalidSegments) {	1✔
246	invalidSegmentNo.add(segment.getSegmentNo());	1✔
247	}	1✔
248	invalidSegmentNo.addAll(segmentsToBeRefreshed);	1✔
249	DistributableDataMapFormat dataMapFormat =	1✔
250	new DistributableDataMapFormat(carbonTable, resolver, validSegments, invalidSegmentNo,
251	partitionsToPrune, false, level, isFallbackJob);	1✔
252	List<ExtendedBlocklet> prunedBlocklets = dataMapJob.execute(dataMapFormat);	1✔
253	// Apply expression on the blocklets.
254	return prunedBlocklets;	1✔
255	}
256
257	public static SegmentStatusManager.ValidAndInvalidSegmentsInfo getValidAndInvalidSegments(
258	CarbonTable carbonTable, Configuration configuration) throws IOException {
259	SegmentStatusManager ssm =	1✔
260	new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier(), configuration);	1✔
261	return ssm.getValidAndInvalidSegments();	1✔
262	}
263
264	/**
265	* Returns valid segment list for a given RelationIdentifier
266	*
267	* @param relationIdentifier get list of segments for relation identifier
268	* @return list of valid segment id's
269	* @throws IOException
270	*/
271	public static List<String> getMainTableValidSegmentList(RelationIdentifier relationIdentifier)
272	throws IOException {
273	List<String> segmentList = new ArrayList<>();	×
274	List<Segment> validSegments = new SegmentStatusManager(AbsoluteTableIdentifier	×
275	.from(relationIdentifier.getTablePath(), relationIdentifier.getDatabaseName(),	×
276	relationIdentifier.getTableName())).getValidAndInvalidSegments().getValidSegments();	×
277	for (Segment segment : validSegments) {	×
278	segmentList.add(segment.getSegmentNo());	×
279	}	×
280	return segmentList;	×
281	}
282
283	}

apache / carbondata / 11940

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