3375

Build # 3375

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/84ce094f2667758cb6820e71a332a1a259db3079">84ce094f2<a href="https://github.com/apache/carbondata/commit/84ce094f2667758cb6820e71a332a1a259db3079">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/85c000e91a1dc84784088f3dc2bc5a13b138344a">85c000e91</a>">85c000e91</a><a href="https://github.com/apache/carbondata/commit/84ce094f2667758cb6820e71a332a1a259db3079"> into 24753a924">24753a924</a>

Run Details

908 of 908 new or added lines in 42 files covered. (100.0%)

63934 of 81095 relevant lines covered (78.84%)

1.04 hits per line

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

44.79

/core/src/main/java/org/apache/carbondata/core/datamap/DistributableDataMapFormat.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.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.io.Serializable;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

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.datastore.block.SegmentPropertiesAndSchemaHolder;
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.schema.table.CarbonTable;
import org.apache.carbondata.core.readcommitter.ReadCommittedScope;
import org.apache.carbondata.core.readcommitter.TableStatusReadCommittedScope;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.util.ObjectSerializationUtil;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.log4j.Logger;

/**
 * Input format for datamaps, it makes the datamap pruning distributable.
 */
public class DistributableDataMapFormat extends FileInputFormat<Void, ExtendedBlocklet>
    implements Serializable, Writable {

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

  private static final long serialVersionUID = 9189779090091151248L;

  private CarbonTable table;

  private FilterResolverIntf filterResolverIntf;

  private List<Segment> validSegments;

  private List<String> invalidSegments;

  private List<PartitionSpec> partitions;

  private boolean isJobToClearDataMaps = false;

  private DataMapLevel dataMapLevel;

  private boolean isFallbackJob = false;

  private String dataMapToClear = "";

  private ReadCommittedScope readCommittedScope;

  DistributableDataMapFormat() {

  }

  DistributableDataMapFormat(CarbonTable table,
      List<Segment> validSegments, List<String> invalidSegments, boolean isJobToClearDataMaps,
      String dataMapToClear) throws IOException {
    this(table, null, validSegments, invalidSegments, null,
        isJobToClearDataMaps, null, false);
    this.dataMapToClear = dataMapToClear;
  }

  DistributableDataMapFormat(CarbonTable table, FilterResolverIntf filterResolverIntf,
      List<Segment> validSegments, List<String> invalidSegments, List<PartitionSpec> partitions,
      boolean isJobToClearDataMaps, DataMapLevel dataMapLevel, boolean isFallbackJob)
      throws IOException {
    this.table = table;
    this.filterResolverIntf = filterResolverIntf;
    this.validSegments = validSegments;
    if (!validSegments.isEmpty()) {
      this.readCommittedScope = validSegments.get(0).getReadCommittedScope();
    }
    this.invalidSegments = invalidSegments;
    this.partitions = partitions;
    this.isJobToClearDataMaps = isJobToClearDataMaps;
    this.dataMapLevel = dataMapLevel;
    this.isFallbackJob = isFallbackJob;
  }

  @Override
  public List<InputSplit> getSplits(JobContext job) throws IOException {
    List<DataMapDistributableWrapper> distributables;
    distributables =
        DataMapChooser.getDefaultDataMap(table, filterResolverIntf).toDistributable(validSegments);
    List<InputSplit> inputSplits = new ArrayList<>(distributables.size());
    inputSplits.addAll(distributables);
    return inputSplits;
  }

  @Override
  public RecordReader<Void, ExtendedBlocklet> createRecordReader(InputSplit inputSplit,
      TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
    return new RecordReader<Void, ExtendedBlocklet>() {
      private Iterator<ExtendedBlocklet> blockletIterator;
      private ExtendedBlocklet currBlocklet;
      private List<DataMap> dataMaps;

      @Override
      public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext)
          throws IOException, InterruptedException {
        DataMapDistributableWrapper distributable = (DataMapDistributableWrapper) inputSplit;
        distributable.getDistributable().getSegment().setCacheable(!isFallbackJob);
        distributable.getDistributable().getSegment().setReadCommittedScope(readCommittedScope);
        List<Segment> segmentsToLoad = new ArrayList<>();
        segmentsToLoad.add(distributable.getDistributable().getSegment());
        if (isJobToClearDataMaps) {
          if (StringUtils.isNotEmpty(dataMapToClear)) {
            List<TableDataMap> dataMaps =
                DataMapStoreManager.getInstance().getAllDataMap(table);
            int i = 0;
            for (TableDataMap tableDataMap : dataMaps) {
              if (tableDataMap != null && dataMapToClear
                  .equalsIgnoreCase(tableDataMap.getDataMapSchema().getDataMapName())) {
                tableDataMap.deleteSegmentDatamapData(
                    ((DataMapDistributableWrapper) inputSplit).getDistributable().getSegment()
                        .getSegmentNo());
                tableDataMap.clear();
                dataMaps.remove(i);
                break;
              }
              i++;
            }
            DataMapStoreManager.getInstance().getAllDataMaps().put(table.getTableUniqueName(),
                dataMaps);
          } else {
            // if job is to clear datamaps just clear datamaps from cache and return
            DataMapStoreManager.getInstance()
                .clearDataMaps(table.getCarbonTableIdentifier().getTableUniqueName());
            // clear the segment properties cache from executor
            SegmentPropertiesAndSchemaHolder.getInstance()
                .invalidate(table.getAbsoluteTableIdentifier());
          }
          List<ExtendedBlocklet> list = new ArrayList<ExtendedBlocklet>();
          list.add(new ExtendedBlocklet());
          blockletIterator = list.iterator();
          return;
        } else if (invalidSegments.size() > 0) {
          // clear the segmentMap and from cache in executor when there are invalid segments
          DataMapStoreManager.getInstance().clearInvalidSegments(table, invalidSegments);
        }
        List<ExtendedBlocklet> blocklets = new ArrayList<>();
        if (dataMapLevel == null) {
          TableDataMap defaultDataMap = DataMapStoreManager.getInstance()
              .getDataMap(table, distributable.getDistributable().getDataMapSchema());
          dataMaps = defaultDataMap.getTableDataMaps(distributable.getDistributable());
          if (table.isTransactionalTable()) {
            blocklets = defaultDataMap.prune(dataMaps, distributable.getDistributable(),
                filterResolverIntf, partitions);
          } else {
            blocklets = defaultDataMap.prune(segmentsToLoad, new DataMapFilter(filterResolverIntf),
                partitions);
          }
          blocklets = DataMapUtil
              .pruneDataMaps(table, filterResolverIntf, segmentsToLoad, partitions, blocklets);
        } else {
          blocklets = DataMapUtil
              .pruneDataMaps(table, filterResolverIntf, segmentsToLoad, partitions, blocklets,
                  dataMapLevel);
        }
        blockletIterator = blocklets.iterator();
      }

      @Override
      public boolean nextKeyValue() throws IOException, InterruptedException {
        boolean hasNext = blockletIterator.hasNext();
        if (hasNext) {
          currBlocklet = blockletIterator.next();
        } else {
          // close all resources when all the results are returned
          close();
        }
        return hasNext;
      }

      @Override
      public Void getCurrentKey() throws IOException, InterruptedException {
        return null;
      }

      @Override
      public ExtendedBlocklet getCurrentValue() throws IOException, InterruptedException {
        return currBlocklet;
      }

      @Override
      public float getProgress() throws IOException, InterruptedException {
        return 0;
      }

      @Override
      public void close() throws IOException {
        if (null != dataMaps) {
          for (DataMap dataMap : dataMaps) {
            dataMap.finish();
          }
        }
      }
    };
  }

  public CarbonTable getCarbonTable() {
    return table;
  }

  @Override
  public void write(DataOutput out) throws IOException {
    table.write(out);
    out.writeInt(invalidSegments.size());
    for (String invalidSegment : invalidSegments) {
      out.writeUTF(invalidSegment);
    }
    out.writeBoolean(isJobToClearDataMaps);
    out.writeBoolean(isFallbackJob);
    if (dataMapLevel == null) {
      out.writeBoolean(false);
    } else {
      out.writeBoolean(true);
      out.writeUTF(dataMapLevel.name());
    }
    out.writeInt(validSegments.size());
    for (Segment segment : validSegments) {
      segment.write(out);
    }
    if (partitions == null) {
      out.writeBoolean(false);
    } else {
      out.writeBoolean(true);
      out.writeInt(partitions.size());
      for (PartitionSpec partitionSpec : partitions) {
        partitionSpec.write(out);
      }
    }
    if (filterResolverIntf != null) {
      out.writeBoolean(true);
      byte[] filterResolverBytes = ObjectSerializationUtil.convertObjectToString(filterResolverIntf)
          .getBytes(Charset.defaultCharset());
      out.writeInt(filterResolverBytes.length);
      out.write(filterResolverBytes);

    } else {
      out.writeBoolean(false);
    }
    out.writeUTF(dataMapToClear);
  }

  @Override
  public void readFields(DataInput in) throws IOException {
    this.table = new CarbonTable();
    table.readFields(in);
    int invalidSegmentSize = in.readInt();
    invalidSegments = new ArrayList<>(invalidSegmentSize);
    for (int i = 0; i < invalidSegmentSize; i++) {
      invalidSegments.add(in.readUTF());
    }
    this.isJobToClearDataMaps = in.readBoolean();
    this.isFallbackJob = in.readBoolean();
    if (in.readBoolean()) {
      this.dataMapLevel = DataMapLevel.valueOf(in.readUTF());
    }
    int validSegmentSize = in.readInt();
    validSegments = new ArrayList<>(validSegmentSize);
    initReadCommittedScope();
    for (int i = 0; i < validSegmentSize; i++) {
      Segment segment = new Segment();
      segment.setReadCommittedScope(readCommittedScope);
      segment.readFields(in);
      validSegments.add(segment);
    }
    if (in.readBoolean()) {
      int numPartitions = in.readInt();
      partitions = new ArrayList<>(numPartitions);
      for (int i = 0; i < numPartitions; i++) {
        PartitionSpec partitionSpec = new PartitionSpec();
        partitionSpec.readFields(in);
        partitions.add(partitionSpec);
      }
    }
    if (in.readBoolean()) {
      byte[] filterResolverBytes = new byte[in.readInt()];
      in.readFully(filterResolverBytes, 0, filterResolverBytes.length);
      this.filterResolverIntf = (FilterResolverIntf) ObjectSerializationUtil
          .convertStringToObject(new String(filterResolverBytes, Charset.defaultCharset()));
    }
    this.dataMapToClear = in.readUTF();
  }

  private void initReadCommittedScope() throws IOException {
    if (readCommittedScope == null) {
      this.readCommittedScope =
          new TableStatusReadCommittedScope(table.getAbsoluteTableIdentifier(),
              FileFactory.getConfiguration());
    }
  }

  /**
   * @return Whether the job is fallback or not.
   */
  public boolean isFallbackJob() {
    return isFallbackJob;
  }

  /**
   * @return Whether the job is to clear cached datamaps or not.
   */
  public boolean isJobToClearDataMaps() {
    return isJobToClearDataMaps;
  }
}

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.datamap;
18
19	import java.io.DataInput;
20	import java.io.DataOutput;
21	import java.io.IOException;
22	import java.io.Serializable;
23	import java.nio.charset.Charset;
24	import java.util.ArrayList;
25	import java.util.Iterator;
26	import java.util.List;
27
28	import org.apache.carbondata.common.logging.LogServiceFactory;
29	import org.apache.carbondata.core.datamap.dev.DataMap;
30	import org.apache.carbondata.core.datamap.dev.expr.DataMapDistributableWrapper;
31	import org.apache.carbondata.core.datastore.block.SegmentPropertiesAndSchemaHolder;
32	import org.apache.carbondata.core.datastore.impl.FileFactory;
33	import org.apache.carbondata.core.indexstore.ExtendedBlocklet;
34	import org.apache.carbondata.core.indexstore.PartitionSpec;
35	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
36	import org.apache.carbondata.core.readcommitter.ReadCommittedScope;
37	import org.apache.carbondata.core.readcommitter.TableStatusReadCommittedScope;
38	import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
39	import org.apache.carbondata.core.util.ObjectSerializationUtil;
40
41	import org.apache.commons.lang.StringUtils;
42	import org.apache.hadoop.io.Writable;
43	import org.apache.hadoop.mapreduce.InputSplit;
44	import org.apache.hadoop.mapreduce.JobContext;
45	import org.apache.hadoop.mapreduce.RecordReader;
46	import org.apache.hadoop.mapreduce.TaskAttemptContext;
47	import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
48	import org.apache.log4j.Logger;
49
50	/**
51	* Input format for datamaps, it makes the datamap pruning distributable.
52	*/
53	public class DistributableDataMapFormat extends FileInputFormat<Void, ExtendedBlocklet>
54	implements Serializable, Writable {
55
56	private static final transient Logger LOGGER =	1✔
57	LogServiceFactory.getLogService(DistributableDataMapFormat.class.getName());	1✔
58
59	private static final long serialVersionUID = 9189779090091151248L;
60
61	private CarbonTable table;
62
63	private FilterResolverIntf filterResolverIntf;
64
65	private List<Segment> validSegments;
66
67	private List<String> invalidSegments;
68
69	private List<PartitionSpec> partitions;
70
71	private boolean isJobToClearDataMaps = false;	1✔
72
73	private DataMapLevel dataMapLevel;
74
75	private boolean isFallbackJob = false;	1✔
76
77	private String dataMapToClear = "";	1✔
78
79	private ReadCommittedScope readCommittedScope;
80
81	DistributableDataMapFormat() {	×
82
83	}	×
84
85	DistributableDataMapFormat(CarbonTable table,
86	List<Segment> validSegments, List<String> invalidSegments, boolean isJobToClearDataMaps,
87	String dataMapToClear) throws IOException {
88	this(table, null, validSegments, invalidSegments, null,	1✔
89	isJobToClearDataMaps, null, false);
90	this.dataMapToClear = dataMapToClear;	1✔
91	}	1✔
92
93	DistributableDataMapFormat(CarbonTable table, FilterResolverIntf filterResolverIntf,
94	List<Segment> validSegments, List<String> invalidSegments, List<PartitionSpec> partitions,
95	boolean isJobToClearDataMaps, DataMapLevel dataMapLevel, boolean isFallbackJob)
96	throws IOException {	1✔
97	this.table = table;	1✔
98	this.filterResolverIntf = filterResolverIntf;	1✔
99	this.validSegments = validSegments;	1✔
100	if (!validSegments.isEmpty()) {	1✔
101	this.readCommittedScope = validSegments.get(0).getReadCommittedScope();	1✔
102	}
103	this.invalidSegments = invalidSegments;	1✔
104	this.partitions = partitions;	1✔
105	this.isJobToClearDataMaps = isJobToClearDataMaps;	1✔
106	this.dataMapLevel = dataMapLevel;	1✔
107	this.isFallbackJob = isFallbackJob;	1✔
108	}	1✔
109
110	@Override
111	public List<InputSplit> getSplits(JobContext job) throws IOException {
112	List<DataMapDistributableWrapper> distributables;
113	distributables =	1✔
114	DataMapChooser.getDefaultDataMap(table, filterResolverIntf).toDistributable(validSegments);	1✔
115	List<InputSplit> inputSplits = new ArrayList<>(distributables.size());	1✔
116	inputSplits.addAll(distributables);	1✔
117	return inputSplits;	1✔
118	}
119
120	@Override
121	public RecordReader<Void, ExtendedBlocklet> createRecordReader(InputSplit inputSplit,
122	TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
123	return new RecordReader<Void, ExtendedBlocklet>() {	1✔
124	private Iterator<ExtendedBlocklet> blockletIterator;
125	private ExtendedBlocklet currBlocklet;
126	private List<DataMap> dataMaps;
127
128	@Override
129	public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext)
130	throws IOException, InterruptedException {
131	DataMapDistributableWrapper distributable = (DataMapDistributableWrapper) inputSplit;	1✔
132	distributable.getDistributable().getSegment().setCacheable(!isFallbackJob);	1✔
133	distributable.getDistributable().getSegment().setReadCommittedScope(readCommittedScope);	1✔
134	List<Segment> segmentsToLoad = new ArrayList<>();	1✔
135	segmentsToLoad.add(distributable.getDistributable().getSegment());	1✔
136	if (isJobToClearDataMaps) {	1✔
137	if (StringUtils.isNotEmpty(dataMapToClear)) {	1✔
138	List<TableDataMap> dataMaps =
139	DataMapStoreManager.getInstance().getAllDataMap(table);	1✔
140	int i = 0;	1✔
141	for (TableDataMap tableDataMap : dataMaps) {	1✔
142	if (tableDataMap != null && dataMapToClear	1✔
143	.equalsIgnoreCase(tableDataMap.getDataMapSchema().getDataMapName())) {	1✔
144	tableDataMap.deleteSegmentDatamapData(	1✔
145	((DataMapDistributableWrapper) inputSplit).getDistributable().getSegment()	1✔
146	.getSegmentNo());	1✔
147	tableDataMap.clear();	1✔
148	dataMaps.remove(i);	1✔
149	break;	1✔
150	}
151	i++;	×
152	}	×
153	DataMapStoreManager.getInstance().getAllDataMaps().put(table.getTableUniqueName(),	1✔
154	dataMaps);
155	} else {	1✔
156	// if job is to clear datamaps just clear datamaps from cache and return
157	DataMapStoreManager.getInstance()	1✔
158	.clearDataMaps(table.getCarbonTableIdentifier().getTableUniqueName());	1✔
159	// clear the segment properties cache from executor
160	SegmentPropertiesAndSchemaHolder.getInstance()	1✔
161	.invalidate(table.getAbsoluteTableIdentifier());	1✔
162	}
163	List<ExtendedBlocklet> list = new ArrayList<ExtendedBlocklet>();	1✔
164	list.add(new ExtendedBlocklet());	1✔
165	blockletIterator = list.iterator();	1✔
166	return;	1✔
167	} else if (invalidSegments.size() > 0) {	1✔
168	// clear the segmentMap and from cache in executor when there are invalid segments
169	DataMapStoreManager.getInstance().clearInvalidSegments(table, invalidSegments);	1✔
170	}
171	List<ExtendedBlocklet> blocklets = new ArrayList<>();	1✔
172	if (dataMapLevel == null) {	1✔
173	TableDataMap defaultDataMap = DataMapStoreManager.getInstance()	×
174	.getDataMap(table, distributable.getDistributable().getDataMapSchema());	×
175	dataMaps = defaultDataMap.getTableDataMaps(distributable.getDistributable());	×
176	if (table.isTransactionalTable()) {	×
177	blocklets = defaultDataMap.prune(dataMaps, distributable.getDistributable(),	×
178	filterResolverIntf, partitions);	×
179	} else {
180	blocklets = defaultDataMap.prune(segmentsToLoad, new DataMapFilter(filterResolverIntf),	×
181	partitions);	×
182	}
183	blocklets = DataMapUtil	×
184	.pruneDataMaps(table, filterResolverIntf, segmentsToLoad, partitions, blocklets);	×
185	} else {	×
186	blocklets = DataMapUtil	1✔
187	.pruneDataMaps(table, filterResolverIntf, segmentsToLoad, partitions, blocklets,	1✔
188	dataMapLevel);	1✔
189	}
190	blockletIterator = blocklets.iterator();	1✔
191	}	1✔
192
193	@Override
194	public boolean nextKeyValue() throws IOException, InterruptedException {
195	boolean hasNext = blockletIterator.hasNext();	1✔
196	if (hasNext) {	1✔
197	currBlocklet = blockletIterator.next();	1✔
198	} else {
199	// close all resources when all the results are returned
200	close();	1✔
201	}
202	return hasNext;	1✔
203	}
204
205	@Override
206	public Void getCurrentKey() throws IOException, InterruptedException {
207	return null;	×
208	}
209
210	@Override
211	public ExtendedBlocklet getCurrentValue() throws IOException, InterruptedException {
212	return currBlocklet;	1✔
213	}
214
215	@Override
216	public float getProgress() throws IOException, InterruptedException {
217	return 0;	×
218	}
219
220	@Override
221	public void close() throws IOException {
222	if (null != dataMaps) {	1✔
223	for (DataMap dataMap : dataMaps) {	×
224	dataMap.finish();	×
225	}	×
226	}
227	}	1✔
228	};
229	}
230
231	public CarbonTable getCarbonTable() {
232	return table;	1✔
233	}
234
235	@Override
236	public void write(DataOutput out) throws IOException {
237	table.write(out);	×
238	out.writeInt(invalidSegments.size());	×
239	for (String invalidSegment : invalidSegments) {	×
240	out.writeUTF(invalidSegment);	×
241	}	×
242	out.writeBoolean(isJobToClearDataMaps);	×
243	out.writeBoolean(isFallbackJob);	×
244	if (dataMapLevel == null) {	×
245	out.writeBoolean(false);	×
246	} else {
247	out.writeBoolean(true);	×
248	out.writeUTF(dataMapLevel.name());	×
249	}
250	out.writeInt(validSegments.size());	×
251	for (Segment segment : validSegments) {	×
252	segment.write(out);	×
253	}	×
254	if (partitions == null) {	×
255	out.writeBoolean(false);	×
256	} else {
257	out.writeBoolean(true);	×
258	out.writeInt(partitions.size());	×
259	for (PartitionSpec partitionSpec : partitions) {	×
260	partitionSpec.write(out);	×
261	}	×
262	}
263	if (filterResolverIntf != null) {	×
264	out.writeBoolean(true);	×
265	byte[] filterResolverBytes = ObjectSerializationUtil.convertObjectToString(filterResolverIntf)	×
266	.getBytes(Charset.defaultCharset());	×
267	out.writeInt(filterResolverBytes.length);	×
268	out.write(filterResolverBytes);	×
269
270	} else {	×
271	out.writeBoolean(false);	×
272	}
273	out.writeUTF(dataMapToClear);	×
274	}	×
275
276	@Override
277	public void readFields(DataInput in) throws IOException {
278	this.table = new CarbonTable();	×
279	table.readFields(in);	×
280	int invalidSegmentSize = in.readInt();	×
281	invalidSegments = new ArrayList<>(invalidSegmentSize);	×
282	for (int i = 0; i < invalidSegmentSize; i++) {	×
283	invalidSegments.add(in.readUTF());	×
284	}
285	this.isJobToClearDataMaps = in.readBoolean();	×
286	this.isFallbackJob = in.readBoolean();	×
287	if (in.readBoolean()) {	×
288	this.dataMapLevel = DataMapLevel.valueOf(in.readUTF());	×
289	}
290	int validSegmentSize = in.readInt();	×
291	validSegments = new ArrayList<>(validSegmentSize);	×
292	initReadCommittedScope();	×
293	for (int i = 0; i < validSegmentSize; i++) {	×
294	Segment segment = new Segment();	×
295	segment.setReadCommittedScope(readCommittedScope);	×
296	segment.readFields(in);	×
297	validSegments.add(segment);	×
298	}
299	if (in.readBoolean()) {	×
300	int numPartitions = in.readInt();	×
301	partitions = new ArrayList<>(numPartitions);	×
302	for (int i = 0; i < numPartitions; i++) {	×
303	PartitionSpec partitionSpec = new PartitionSpec();	×
304	partitionSpec.readFields(in);	×
305	partitions.add(partitionSpec);	×
306	}
307	}
308	if (in.readBoolean()) {	×
309	byte[] filterResolverBytes = new byte[in.readInt()];	×
310	in.readFully(filterResolverBytes, 0, filterResolverBytes.length);	×
311	this.filterResolverIntf = (FilterResolverIntf) ObjectSerializationUtil	×
312	.convertStringToObject(new String(filterResolverBytes, Charset.defaultCharset()));	×
313	}
314	this.dataMapToClear = in.readUTF();	×
315	}	×
316
317	private void initReadCommittedScope() throws IOException {
318	if (readCommittedScope == null) {	×
319	this.readCommittedScope =	×
320	new TableStatusReadCommittedScope(table.getAbsoluteTableIdentifier(),	×
321	FileFactory.getConfiguration());	×
322	}
323	}	×
324
325	/**
326	* @return Whether the job is fallback or not.
327	*/
328	public boolean isFallbackJob() {
329	return isFallbackJob;	1✔
330	}
331
332	/**
333	* @return Whether the job is to clear cached datamaps or not.
334	*/
335	public boolean isJobToClearDataMaps() {
336	return isJobToClearDataMaps;	×
337	}
338	}

apache / carbondata / 3375

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