3825

Build # 3825

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/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/2b3f45bf9b7b38d9c60dc8794cd4c6b113eaa33b">2b3f45bf9">&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/2b3f45bf9b7b38d9c60dc8794cd4c6b113eaa33b">2b3f45bf9</a><a href="https://github.com/apache/carbondata/commit/2b3f45bf9b7b38d9c60dc8794cd4c6b113eaa33b">&lt;a href=&quot;https://github.com/apache/carbondata/commit/2b3f45bf9b7b38d9c60dc8794cd4c6b113eaa33b&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 class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/apache/carbondata/commit/2b3f45bf9b7b38d9c60dc8794cd4c6b113eaa33b&amp;quot;&amp;gt;2b3f45bf9&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/2b3f45bf9b7b38d9c60dc8794cd4c6b113eaa33b&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/2b3f45bf9b7b38d9c60dc8794cd4c6b1&amp;lt;/a&amp;gt;13eaa</a>33b&quot;&gt;&amp;quot;&amp;gt;Merge &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;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/5b7a0f834&quot;&gt;5b7a0f8... (continued)

Run Details

64092 of 82887 relevant lines covered (77.32%)

1.01 hits per line

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

89.58

/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.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.hadoop.api;

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.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.status.DataMapStatusManager;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.indexstore.PartitionSpec;
import org.apache.carbondata.core.locks.CarbonLockFactory;
import org.apache.carbondata.core.locks.ICarbonLock;
import org.apache.carbondata.core.locks.LockUsage;
import org.apache.carbondata.core.metadata.SegmentFileStore;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.mutate.CarbonUpdateUtil;
import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
import org.apache.carbondata.core.statusmanager.SegmentStatus;
import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
import org.apache.carbondata.core.util.CarbonSessionInfo;
import org.apache.carbondata.core.util.ThreadLocalSessionInfo;
import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.events.OperationContext;
import org.apache.carbondata.events.OperationListenerBus;
import org.apache.carbondata.processing.loading.events.LoadEvents;
import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
import org.apache.carbondata.processing.util.CarbonLoaderUtil;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.JobStatus;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.log4j.Logger;

/**
 * Outputcommitter which manages the segments during loading.It commits segment information to the
 * tablestatus file upon success or fail.
 */
public class CarbonOutputCommitter extends FileOutputCommitter {

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

  private ICarbonLock segmentLock;

  public CarbonOutputCommitter(Path outputPath, TaskAttemptContext context) throws IOException {
    super(outputPath, context);
  }

  /**
   * Update the tablestatus with inprogress while setup the job.
   *
   * @param context
   * @throws IOException
   */
  @Override public void setupJob(JobContext context) throws IOException {
    super.setupJob(context);
    boolean overwriteSet = CarbonTableOutputFormat.isOverwriteSet(context.getConfiguration());
    CarbonLoadModel loadModel = CarbonTableOutputFormat.getLoadModel(context.getConfiguration());
    if (loadModel.getSegmentId() == null) {
      CarbonLoaderUtil.readAndUpdateLoadProgressInTableMeta(loadModel, overwriteSet);
    }
    // Take segment lock
    segmentLock = CarbonLockFactory.getCarbonLockObj(
        loadModel.getCarbonDataLoadSchema().getCarbonTable().getAbsoluteTableIdentifier(),
        CarbonTablePath.addSegmentPrefix(loadModel.getSegmentId()) + LockUsage.LOCK);
    if (!segmentLock.lockWithRetries()) {
      throw new RuntimeException("Already segment is locked for loading, not supposed happen");
    }
    CarbonTableOutputFormat.setLoadModel(context.getConfiguration(), loadModel);
  }

  @Override public void setupTask(TaskAttemptContext context) throws IOException {
    super.setupTask(context);
  }

  /**
   * Update the tablestatus as success after job is success
   *
   * @param context
   * @throws IOException
   */
  @Override public void commitJob(JobContext context) throws IOException {
    try {
      super.commitJob(context);
    } catch (IOException e) {
      // ignore, in case of concurrent load it try to remove temporary folders by other load may
      // cause file not found exception. This will not impact carbon load,
      LOGGER.warn(e.getMessage());
    }
    boolean overwriteSet = CarbonTableOutputFormat.isOverwriteSet(context.getConfiguration());
    CarbonLoadModel loadModel = CarbonTableOutputFormat.getLoadModel(context.getConfiguration());
    LoadMetadataDetails newMetaEntry = loadModel.getCurrentLoadMetadataDetail();
    String readPath = CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath())
        + CarbonCommonConstants.FILE_SEPARATOR
        + loadModel.getSegmentId() + "_" + loadModel.getFactTimeStamp() + ".tmp";
    // Merge all partition files into a single file.
    String segmentFileName = SegmentFileStore.genSegmentFileName(
        loadModel.getSegmentId(), String.valueOf(loadModel.getFactTimeStamp()));
    SegmentFileStore.SegmentFile segmentFile = SegmentFileStore
        .mergeSegmentFiles(readPath, segmentFileName,
            CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath()));
    if (segmentFile != null) {
      if (null == newMetaEntry) {
        throw new RuntimeException("Internal Error");
      }
      // Move all files from temp directory of each segment to partition directory
      SegmentFileStore.moveFromTempFolder(segmentFile,
          loadModel.getSegmentId() + "_" + loadModel.getFactTimeStamp() + ".tmp",
          loadModel.getTablePath());
      newMetaEntry.setSegmentFile(segmentFileName + CarbonTablePath.SEGMENT_EXT);
    }
    OperationContext operationContext = (OperationContext) getOperationContext();
    CarbonTable carbonTable = loadModel.getCarbonDataLoadSchema().getCarbonTable();
    String uuid = "";
    if (loadModel.getCarbonDataLoadSchema().getCarbonTable().isChildDataMap() &&
        operationContext != null) {
      uuid = operationContext.getProperty("uuid").toString();
    }

    SegmentFileStore.updateSegmentFile(carbonTable, loadModel.getSegmentId(),
        segmentFileName + CarbonTablePath.SEGMENT_EXT,
        carbonTable.getCarbonTableIdentifier().getTableId(),
        new SegmentFileStore(carbonTable.getTablePath(),
            segmentFileName + CarbonTablePath.SEGMENT_EXT));

    CarbonLoaderUtil
        .populateNewLoadMetaEntry(newMetaEntry, SegmentStatus.SUCCESS, loadModel.getFactTimeStamp(),
            true);
    long segmentSize = CarbonLoaderUtil
        .addDataIndexSizeIntoMetaEntry(newMetaEntry, loadModel.getSegmentId(), carbonTable);
    if (segmentSize > 0 || overwriteSet) {
      if (operationContext != null) {
        operationContext
            .setProperty("current.segmentfile", newMetaEntry.getSegmentFile());
        LoadEvents.LoadTablePreStatusUpdateEvent event =
            new LoadEvents.LoadTablePreStatusUpdateEvent(carbonTable.getCarbonTableIdentifier(),
                loadModel);
        try {
          OperationListenerBus.getInstance().fireEvent(event, operationContext);
        } catch (Exception e) {
          throw new IOException(e);
        }
      }
      String uniqueId = null;
      if (overwriteSet) {
        if (!loadModel.isCarbonTransactionalTable()) {
          CarbonLoaderUtil.deleteNonTransactionalTableForInsertOverwrite(loadModel);
        } else {
          if (segmentSize == 0) {
            newMetaEntry.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);
          }
          uniqueId = overwritePartitions(loadModel, newMetaEntry, uuid);
        }
      } else {
        CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false, uuid);
      }
      DataMapStatusManager.disableAllLazyDataMaps(carbonTable);
      if (operationContext != null) {
        LoadEvents.LoadTablePostStatusUpdateEvent postStatusUpdateEvent =
            new LoadEvents.LoadTablePostStatusUpdateEvent(loadModel);
        try {
          OperationListenerBus.getInstance()
              .fireEvent(postStatusUpdateEvent, operationContext);
        } catch (Exception e) {
          throw new IOException(e);
        }
      }
      String updateTime =
          context.getConfiguration().get(CarbonTableOutputFormat.UPADTE_TIMESTAMP, null);
      String segmentsToBeDeleted =
          context.getConfiguration().get(CarbonTableOutputFormat.SEGMENTS_TO_BE_DELETED, "");
      List<Segment> segmentDeleteList = Segment.toSegmentList(segmentsToBeDeleted.split(","), null);
      Set<Segment> segmentSet = new HashSet<>(
          new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier(),
              context.getConfiguration()).getValidAndInvalidSegments().getValidSegments());
      if (updateTime != null) {
        CarbonUpdateUtil.updateTableMetadataStatus(segmentSet, carbonTable, updateTime, true,
            segmentDeleteList);
      } else if (uniqueId != null) {
        CarbonUpdateUtil.updateTableMetadataStatus(segmentSet, carbonTable, uniqueId, true,
            segmentDeleteList);
      }
    } else {
      CarbonLoaderUtil.updateTableStatusForFailure(loadModel);
    }
    if (segmentLock != null) {
      segmentLock.unlock();
    }
  }

  /**
   * Overwrite the partitions in case of overwrite query. It just updates the partition map files
   * of all segment files.
   *
   * @param loadModel
   * @return
   * @throws IOException
   */
  private String overwritePartitions(CarbonLoadModel loadModel, LoadMetadataDetails newMetaEntry,
      String uuid) throws IOException {
    CarbonTable table = loadModel.getCarbonDataLoadSchema().getCarbonTable();
    SegmentFileStore fileStore = new SegmentFileStore(loadModel.getTablePath(),
        loadModel.getSegmentId() + "_" + loadModel.getFactTimeStamp()
            + CarbonTablePath.SEGMENT_EXT);
    List<PartitionSpec> partitionSpecs = fileStore.getPartitionSpecs();

    if (partitionSpecs != null && partitionSpecs.size() > 0) {
      List<Segment> validSegments =
          new SegmentStatusManager(table.getAbsoluteTableIdentifier())
              .getValidAndInvalidSegments().getValidSegments();
      String uniqueId = String.valueOf(System.currentTimeMillis());
      List<String> tobeUpdatedSegs = new ArrayList<>();
      List<String> tobeDeletedSegs = new ArrayList<>();
      // First drop the partitions from partition mapper files of each segment
      for (Segment segment : validSegments) {
        new SegmentFileStore(table.getTablePath(), segment.getSegmentFileName())
            .dropPartitions(segment, partitionSpecs, uniqueId, tobeDeletedSegs, tobeUpdatedSegs);

      }
      newMetaEntry.setUpdateStatusFileName(uniqueId);
      // Commit the removed partitions in carbon store.
      CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false, uuid,
          Segment.toSegmentList(tobeDeletedSegs, null),
          Segment.toSegmentList(tobeUpdatedSegs, null));
      return uniqueId;
    }
    return null;
  }

  private Object getOperationContext() {
    // when validate segments is disabled in thread local update it to CarbonTableInputFormat
    CarbonSessionInfo carbonSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo();
    if (carbonSessionInfo != null) {
      return carbonSessionInfo.getThreadParams().getExtraInfo("partition.operationcontext");
    }
    return null;
  }

  /**
   * Update the tablestatus as fail if any fail happens.And also clean up the temp folders if any
   * are existed.
   *
   * @param context
   * @param state
   * @throws IOException
   */
  @Override public void abortJob(JobContext context, JobStatus.State state) throws IOException {
    try {
      super.abortJob(context, state);
      CarbonLoadModel loadModel = CarbonTableOutputFormat.getLoadModel(context.getConfiguration());
      CarbonLoaderUtil.updateTableStatusForFailure(loadModel);
      String segmentFileName = loadModel.getSegmentId() + "_" + loadModel.getFactTimeStamp();
      LoadMetadataDetails metadataDetail = loadModel.getCurrentLoadMetadataDetail();
      if (metadataDetail != null) {
        // In case the segment file is already created for this job then just link it so that it
        // will be used while cleaning.
        if (!metadataDetail.getSegmentStatus().equals(SegmentStatus.SUCCESS)) {
          String readPath = CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath())
              + CarbonCommonConstants.FILE_SEPARATOR + segmentFileName
              + CarbonTablePath.SEGMENT_EXT;
          if (FileFactory.getCarbonFile(readPath).exists()) {
            metadataDetail.setSegmentFile(segmentFileName + CarbonTablePath.SEGMENT_EXT);
          }
        }
      }
      // Clean the temp files
      CarbonFile segTmpFolder = FileFactory.getCarbonFile(
          CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath())
              + CarbonCommonConstants.FILE_SEPARATOR + segmentFileName + ".tmp");
      // delete temp segment folder
      if (segTmpFolder.exists()) {
        FileFactory.deleteAllCarbonFilesOfDir(segTmpFolder);
      }
      CarbonFile segmentFilePath = FileFactory.getCarbonFile(
          CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath())
              + CarbonCommonConstants.FILE_SEPARATOR + segmentFileName
              + CarbonTablePath.SEGMENT_EXT);
      // Delete the temp data folders of this job if exists
      if (segmentFilePath.exists()) {
        SegmentFileStore fileStore = new SegmentFileStore(loadModel.getTablePath(),
            segmentFileName + CarbonTablePath.SEGMENT_EXT);
        SegmentFileStore.removeTempFolder(fileStore.getLocationMap(), segmentFileName + ".tmp",
            loadModel.getTablePath());
      }
      LOGGER.error("Loading failed with job status : " + state);
    } finally {
      if (segmentLock != null) {
        segmentLock.unlock();
      }
    }
  }

}

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.hadoop.api;
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.constants.CarbonCommonConstants;
28	import org.apache.carbondata.core.datamap.Segment;
29	import org.apache.carbondata.core.datamap.status.DataMapStatusManager;
30	import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
31	import org.apache.carbondata.core.datastore.impl.FileFactory;
32	import org.apache.carbondata.core.indexstore.PartitionSpec;
33	import org.apache.carbondata.core.locks.CarbonLockFactory;
34	import org.apache.carbondata.core.locks.ICarbonLock;
35	import org.apache.carbondata.core.locks.LockUsage;
36	import org.apache.carbondata.core.metadata.SegmentFileStore;
37	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
38	import org.apache.carbondata.core.mutate.CarbonUpdateUtil;
39	import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
40	import org.apache.carbondata.core.statusmanager.SegmentStatus;
41	import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
42	import org.apache.carbondata.core.util.CarbonSessionInfo;
43	import org.apache.carbondata.core.util.ThreadLocalSessionInfo;
44	import org.apache.carbondata.core.util.path.CarbonTablePath;
45	import org.apache.carbondata.events.OperationContext;
46	import org.apache.carbondata.events.OperationListenerBus;
47	import org.apache.carbondata.processing.loading.events.LoadEvents;
48	import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
49	import org.apache.carbondata.processing.util.CarbonLoaderUtil;
50
51	import org.apache.hadoop.fs.Path;
52	import org.apache.hadoop.mapreduce.JobContext;
53	import org.apache.hadoop.mapreduce.JobStatus;
54	import org.apache.hadoop.mapreduce.TaskAttemptContext;
55	import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
56	import org.apache.log4j.Logger;
57
58	/**
59	* Outputcommitter which manages the segments during loading.It commits segment information to the
60	* tablestatus file upon success or fail.
61	*/
62	public class CarbonOutputCommitter extends FileOutputCommitter {
63
64	private static final Logger LOGGER =	2✔
65	LogServiceFactory.getLogService(CarbonOutputCommitter.class.getName());	2✔
66
67	private ICarbonLock segmentLock;
68
69	public CarbonOutputCommitter(Path outputPath, TaskAttemptContext context) throws IOException {
70	super(outputPath, context);	2✔
71	}	2✔
72
73	/**
74	* Update the tablestatus with inprogress while setup the job.
75	*
76	* @param context
77	* @throws IOException
78	*/
79	@Override public void setupJob(JobContext context) throws IOException {
80	super.setupJob(context);	2✔
81	boolean overwriteSet = CarbonTableOutputFormat.isOverwriteSet(context.getConfiguration());	2✔
82	CarbonLoadModel loadModel = CarbonTableOutputFormat.getLoadModel(context.getConfiguration());	2✔
83	if (loadModel.getSegmentId() == null) {	2✔
84	CarbonLoaderUtil.readAndUpdateLoadProgressInTableMeta(loadModel, overwriteSet);	×
85	}
86	// Take segment lock
87	segmentLock = CarbonLockFactory.getCarbonLockObj(	2✔
88	loadModel.getCarbonDataLoadSchema().getCarbonTable().getAbsoluteTableIdentifier(),	2✔
89	CarbonTablePath.addSegmentPrefix(loadModel.getSegmentId()) + LockUsage.LOCK);	2✔
90	if (!segmentLock.lockWithRetries()) {	2✔
91	throw new RuntimeException("Already segment is locked for loading, not supposed happen");	×
92	}
93	CarbonTableOutputFormat.setLoadModel(context.getConfiguration(), loadModel);	2✔
94	}	2✔
95
96	@Override public void setupTask(TaskAttemptContext context) throws IOException {
97	super.setupTask(context);	2✔
98	}	2✔
99
100	/**
101	* Update the tablestatus as success after job is success
102	*
103	* @param context
104	* @throws IOException
105	*/
106	@Override public void commitJob(JobContext context) throws IOException {
107	try {
108	super.commitJob(context);	2✔
109	} catch (IOException e) {	1✔
110	// ignore, in case of concurrent load it try to remove temporary folders by other load may
111	// cause file not found exception. This will not impact carbon load,
112	LOGGER.warn(e.getMessage());	1✔
113	}	2✔
114	boolean overwriteSet = CarbonTableOutputFormat.isOverwriteSet(context.getConfiguration());	2✔
115	CarbonLoadModel loadModel = CarbonTableOutputFormat.getLoadModel(context.getConfiguration());	2✔
116	LoadMetadataDetails newMetaEntry = loadModel.getCurrentLoadMetadataDetail();	2✔
117	String readPath = CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath())	2✔
118	+ CarbonCommonConstants.FILE_SEPARATOR
119	+ loadModel.getSegmentId() + "_" + loadModel.getFactTimeStamp() + ".tmp";	2✔
120	// Merge all partition files into a single file.
121	String segmentFileName = SegmentFileStore.genSegmentFileName(	2✔
122	loadModel.getSegmentId(), String.valueOf(loadModel.getFactTimeStamp()));	2✔
123	SegmentFileStore.SegmentFile segmentFile = SegmentFileStore	2✔
124	.mergeSegmentFiles(readPath, segmentFileName,	2✔
125	CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath()));	2✔
126	if (segmentFile != null) {	2✔
127	if (null == newMetaEntry) {	1✔
128	throw new RuntimeException("Internal Error");	×
129	}
130	// Move all files from temp directory of each segment to partition directory
131	SegmentFileStore.moveFromTempFolder(segmentFile,	1✔
132	loadModel.getSegmentId() + "_" + loadModel.getFactTimeStamp() + ".tmp",	1✔
133	loadModel.getTablePath());	1✔
134	newMetaEntry.setSegmentFile(segmentFileName + CarbonTablePath.SEGMENT_EXT);	1✔
135	}
136	OperationContext operationContext = (OperationContext) getOperationContext();	2✔
137	CarbonTable carbonTable = loadModel.getCarbonDataLoadSchema().getCarbonTable();	2✔
138	String uuid = "";	2✔
139	if (loadModel.getCarbonDataLoadSchema().getCarbonTable().isChildDataMap() &&	2✔
140	operationContext != null) {
141	uuid = operationContext.getProperty("uuid").toString();	1✔
142	}
143
144	SegmentFileStore.updateSegmentFile(carbonTable, loadModel.getSegmentId(),	2✔
145	segmentFileName + CarbonTablePath.SEGMENT_EXT,
146	carbonTable.getCarbonTableIdentifier().getTableId(),	2✔
147	new SegmentFileStore(carbonTable.getTablePath(),	2✔
148	segmentFileName + CarbonTablePath.SEGMENT_EXT));
149
150	CarbonLoaderUtil	2✔
151	.populateNewLoadMetaEntry(newMetaEntry, SegmentStatus.SUCCESS, loadModel.getFactTimeStamp(),	1✔
152	true);
153	long segmentSize = CarbonLoaderUtil	1✔
154	.addDataIndexSizeIntoMetaEntry(newMetaEntry, loadModel.getSegmentId(), carbonTable);	1✔
155	if (segmentSize > 0 \|\| overwriteSet) {	1✔
156	if (operationContext != null) {	1✔
157	operationContext	1✔
158	.setProperty("current.segmentfile", newMetaEntry.getSegmentFile());	1✔
159	LoadEvents.LoadTablePreStatusUpdateEvent event =	1✔
160	new LoadEvents.LoadTablePreStatusUpdateEvent(carbonTable.getCarbonTableIdentifier(),	1✔
161	loadModel);
162	try {
163	OperationListenerBus.getInstance().fireEvent(event, operationContext);	1✔
164	} catch (Exception e) {	×
165	throw new IOException(e);	×
166	}	1✔
167	}
168	String uniqueId = null;	1✔
169	if (overwriteSet) {	1✔
170	if (!loadModel.isCarbonTransactionalTable()) {	1✔
171	CarbonLoaderUtil.deleteNonTransactionalTableForInsertOverwrite(loadModel);	×
172	} else {
173	if (segmentSize == 0) {	1✔
174	newMetaEntry.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);	1✔
175	}
176	uniqueId = overwritePartitions(loadModel, newMetaEntry, uuid);	1✔
177	}
178	} else {
179	CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false, uuid);	1✔
180	}
181	DataMapStatusManager.disableAllLazyDataMaps(carbonTable);	1✔
182	if (operationContext != null) {	1✔
183	LoadEvents.LoadTablePostStatusUpdateEvent postStatusUpdateEvent =	1✔
184	new LoadEvents.LoadTablePostStatusUpdateEvent(loadModel);
185	try {
186	OperationListenerBus.getInstance()	1✔
187	.fireEvent(postStatusUpdateEvent, operationContext);	1✔
188	} catch (Exception e) {	×
189	throw new IOException(e);	×
190	}	1✔
191	}
192	String updateTime =	1✔
193	context.getConfiguration().get(CarbonTableOutputFormat.UPADTE_TIMESTAMP, null);	1✔
194	String segmentsToBeDeleted =	1✔
195	context.getConfiguration().get(CarbonTableOutputFormat.SEGMENTS_TO_BE_DELETED, "");	1✔
196	List<Segment> segmentDeleteList = Segment.toSegmentList(segmentsToBeDeleted.split(","), null);	1✔
197	Set<Segment> segmentSet = new HashSet<>(	1✔
198	new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier(),	1✔
199	context.getConfiguration()).getValidAndInvalidSegments().getValidSegments());	1✔
200	if (updateTime != null) {	1✔
201	CarbonUpdateUtil.updateTableMetadataStatus(segmentSet, carbonTable, updateTime, true,	1✔
202	segmentDeleteList);
203	} else if (uniqueId != null) {	1✔
204	CarbonUpdateUtil.updateTableMetadataStatus(segmentSet, carbonTable, uniqueId, true,	1✔
205	segmentDeleteList);
206	}
207	} else {	1✔
208	CarbonLoaderUtil.updateTableStatusForFailure(loadModel);	1✔
209	}
210	if (segmentLock != null) {	1✔
211	segmentLock.unlock();	1✔
212	}
213	}	1✔
214
215	/**
216	* Overwrite the partitions in case of overwrite query. It just updates the partition map files
217	* of all segment files.
218	*
219	* @param loadModel
220	* @return
221	* @throws IOException
222	*/
223	private String overwritePartitions(CarbonLoadModel loadModel, LoadMetadataDetails newMetaEntry,
224	String uuid) throws IOException {
225	CarbonTable table = loadModel.getCarbonDataLoadSchema().getCarbonTable();	1✔
226	SegmentFileStore fileStore = new SegmentFileStore(loadModel.getTablePath(),	1✔
227	loadModel.getSegmentId() + "_" + loadModel.getFactTimeStamp()	1✔
228	+ CarbonTablePath.SEGMENT_EXT);
229	List<PartitionSpec> partitionSpecs = fileStore.getPartitionSpecs();	1✔
230
231	if (partitionSpecs != null && partitionSpecs.size() > 0) {	1✔
232	List<Segment> validSegments =	1✔
233	new SegmentStatusManager(table.getAbsoluteTableIdentifier())	1✔
234	.getValidAndInvalidSegments().getValidSegments();	1✔
235	String uniqueId = String.valueOf(System.currentTimeMillis());	1✔
236	List<String> tobeUpdatedSegs = new ArrayList<>();	1✔
237	List<String> tobeDeletedSegs = new ArrayList<>();	1✔
238	// First drop the partitions from partition mapper files of each segment
239	for (Segment segment : validSegments) {	1✔
240	new SegmentFileStore(table.getTablePath(), segment.getSegmentFileName())	1✔
241	.dropPartitions(segment, partitionSpecs, uniqueId, tobeDeletedSegs, tobeUpdatedSegs);	1✔
242
243	}	1✔
244	newMetaEntry.setUpdateStatusFileName(uniqueId);	1✔
245	// Commit the removed partitions in carbon store.
246	CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false, uuid,	1✔
247	Segment.toSegmentList(tobeDeletedSegs, null),	1✔
248	Segment.toSegmentList(tobeUpdatedSegs, null));	1✔
249	return uniqueId;	1✔
250	}
251	return null;	1✔
252	}
253
254	private Object getOperationContext() {
255	// when validate segments is disabled in thread local update it to CarbonTableInputFormat
256	CarbonSessionInfo carbonSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo();	2✔
257	if (carbonSessionInfo != null) {	2✔
258	return carbonSessionInfo.getThreadParams().getExtraInfo("partition.operationcontext");	2✔
259	}
260	return null;	×
261	}
262
263	/**
264	* Update the tablestatus as fail if any fail happens.And also clean up the temp folders if any
265	* are existed.
266	*
267	* @param context
268	* @param state
269	* @throws IOException
270	*/
271	@Override public void abortJob(JobContext context, JobStatus.State state) throws IOException {
272	try {
273	super.abortJob(context, state);	2✔
274	CarbonLoadModel loadModel = CarbonTableOutputFormat.getLoadModel(context.getConfiguration());	2✔
275	CarbonLoaderUtil.updateTableStatusForFailure(loadModel);	2✔
276	String segmentFileName = loadModel.getSegmentId() + "_" + loadModel.getFactTimeStamp();	2✔
277	LoadMetadataDetails metadataDetail = loadModel.getCurrentLoadMetadataDetail();	2✔
278	if (metadataDetail != null) {	2✔
279	// In case the segment file is already created for this job then just link it so that it
280	// will be used while cleaning.
281	if (!metadataDetail.getSegmentStatus().equals(SegmentStatus.SUCCESS)) {	1✔
282	String readPath = CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath())	1✔
283	+ CarbonCommonConstants.FILE_SEPARATOR + segmentFileName
284	+ CarbonTablePath.SEGMENT_EXT;
285	if (FileFactory.getCarbonFile(readPath).exists()) {	1✔
286	metadataDetail.setSegmentFile(segmentFileName + CarbonTablePath.SEGMENT_EXT);	×
287	}
288	}
289	}
290	// Clean the temp files
291	CarbonFile segTmpFolder = FileFactory.getCarbonFile(	2✔
292	CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath())	2✔
293	+ CarbonCommonConstants.FILE_SEPARATOR + segmentFileName + ".tmp");
294	// delete temp segment folder
295	if (segTmpFolder.exists()) {	2✔
296	FileFactory.deleteAllCarbonFilesOfDir(segTmpFolder);	×
297	}
298	CarbonFile segmentFilePath = FileFactory.getCarbonFile(	2✔
299	CarbonTablePath.getSegmentFilesLocation(loadModel.getTablePath())	2✔
300	+ CarbonCommonConstants.FILE_SEPARATOR + segmentFileName
301	+ CarbonTablePath.SEGMENT_EXT);
302	// Delete the temp data folders of this job if exists
303	if (segmentFilePath.exists()) {	2✔
304	SegmentFileStore fileStore = new SegmentFileStore(loadModel.getTablePath(),	×
305	segmentFileName + CarbonTablePath.SEGMENT_EXT);
306	SegmentFileStore.removeTempFolder(fileStore.getLocationMap(), segmentFileName + ".tmp",	×
307	loadModel.getTablePath());	×
308	}
309	LOGGER.error("Loading failed with job status : " + state);	2✔
310	} finally {
311	if (segmentLock != null) {	2✔
312	segmentLock.unlock();	2✔
313	}
314	}	×
315	}	2✔
316
317	}

apache / carbondata / 3825

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