1748

Build # 1748

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/2479b74dac617ea9b0b57524c260e48cc786a5b0">2479b74da<a href="https://github.com/apache/carbondata/commit/2479b74dac617ea9b0b57524c260e48cc786a5b0">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/f192985aeb77d6e5f359b69805533c52ad8a97d9">f192985ae</a>">f192985ae</a><a href="https://github.com/apache/carbondata/commit/2479b74dac617ea9b0b57524c260e48cc786a5b0"> into f51c40add">f51c40add</a>

Run Details

42 of 42 new or added lines in 1 file covered. (100.0%)

61557 of 77393 relevant lines covered (79.54%)

1.05 hits per line

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

87.27

/processing/src/main/java/org/apache/carbondata/processing/loading/TableProcessingOperations.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.processing.loading;

import java.io.File;
import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.core.util.CarbonThreadFactory;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
import org.apache.carbondata.processing.util.CarbonLoaderUtil;

import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;

public class TableProcessingOperations {
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(CarbonLoaderUtil.class.getName());

  /**
   *
   * @param carbonTable
   * @param isCompactionFlow
   * @throws IOException
   */
  public static void deletePartialLoadDataIfExist(CarbonTable carbonTable,
      final boolean isCompactionFlow) throws IOException {
    String metaDataLocation = carbonTable.getMetadataPath();
    final LoadMetadataDetails[] details = SegmentStatusManager.readLoadMetadata(metaDataLocation);

    //delete folder which metadata no exist in tablestatus
    String partitionPath = CarbonTablePath.getPartitionDir(carbonTable.getTablePath());
    FileFactory.FileType fileType = FileFactory.getFileType(partitionPath);
    if (FileFactory.isFileExist(partitionPath, fileType)) {
      CarbonFile carbonFile = FileFactory.getCarbonFile(partitionPath, fileType);
      CarbonFile[] listFiles = carbonFile.listFiles(new CarbonFileFilter() {
        @Override public boolean accept(CarbonFile path) {
          String segmentId =
              CarbonTablePath.DataFileUtil.getSegmentIdFromPath(path.getAbsolutePath() + "/dummy");
          boolean found = false;
          for (int j = 0; j < details.length; j++) {
            if (details[j].getLoadName().equals(segmentId)) {
              found = true;
              break;
            }
          }
          return !found;
        }
      });
      for (int k = 0; k < listFiles.length; k++) {
        String segmentId = CarbonTablePath.DataFileUtil
            .getSegmentIdFromPath(listFiles[k].getAbsolutePath() + "/dummy");
        if (isCompactionFlow) {
          if (segmentId.contains(".")) {
            CarbonLoaderUtil.deleteStorePath(listFiles[k].getAbsolutePath());
          }
        } else {
          if (!segmentId.contains(".")) {
            CarbonLoaderUtil.deleteStorePath(listFiles[k].getAbsolutePath());
          }
        }
      }
    }
  }

  /**
   *
   * This method will delete the local data load folder location after data load is complete
   *
   * @param loadModel
   * @param isCompactionFlow COMPACTION keyword will be added to path to make path unique if true
   * @param isAltPartitionFlow Alter_Partition keyword will be added to path to make path unique if
   *                           true
   */
  public static void deleteLocalDataLoadFolderLocation(CarbonLoadModel loadModel,
      boolean isCompactionFlow, boolean isAltPartitionFlow) {
    String tableName = loadModel.getTableName();
    String databaseName = loadModel.getDatabaseName();
    String tempLocationKey = CarbonDataProcessorUtil
        .getTempStoreLocationKey(databaseName, tableName, loadModel.getSegmentId(),
            loadModel.getTaskNo(), isCompactionFlow, isAltPartitionFlow);
    deleteLocalDataLoadFolderLocation(tempLocationKey, tableName);
  }

  /**
   *
   * This method will delete the local data load folder location after data load is complete
   *
   * @param tempLocationKey temporary location set in carbon properties
   * @param tableName
   */
  public static void deleteLocalDataLoadFolderLocation(String tempLocationKey, String tableName) {

    // form local store location
    final String localStoreLocations = CarbonProperties.getInstance().getProperty(tempLocationKey);
    if (localStoreLocations == null) {
      throw new RuntimeException("Store location not set for the key " + tempLocationKey);
    }
    // submit local folder clean up in another thread so that main thread execution is not blocked
    ExecutorService localFolderDeletionService = Executors
        .newFixedThreadPool(1, new CarbonThreadFactory("LocalFolderDeletionPool:" + tableName));
    try {
      localFolderDeletionService.submit(new Callable<Void>() {
        @Override public Void call() throws Exception {
          long startTime = System.currentTimeMillis();
          String[] locArray = StringUtils.split(localStoreLocations, File.pathSeparator);
          for (String loc : locArray) {
            try {
              CarbonUtil.deleteFoldersAndFiles(new File(loc));
            } catch (IOException | InterruptedException e) {
              LOGGER.error("Failed to delete local data load folder location: " + loc, e);
            }
          }
          LOGGER.info(
              "Deleted the local store location: " + localStoreLocations + " : Time taken: " + (
                  System.currentTimeMillis() - startTime));
          return null;
        }
      });
    } finally {
      CarbonProperties.getInstance().removeProperty(tempLocationKey);
      if (null != localFolderDeletionService) {
        localFolderDeletionService.shutdown();
      }
    }

  }
}

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.processing.loading;
19
20	import java.io.File;
21	import java.io.IOException;
22	import java.util.concurrent.Callable;
23	import java.util.concurrent.ExecutorService;
24	import java.util.concurrent.Executors;
25
26	import org.apache.carbondata.common.logging.LogServiceFactory;
27	import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
28	import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
29	import org.apache.carbondata.core.datastore.impl.FileFactory;
30	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
31	import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
32	import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
33	import org.apache.carbondata.core.util.CarbonProperties;
34	import org.apache.carbondata.core.util.CarbonThreadFactory;
35	import org.apache.carbondata.core.util.CarbonUtil;
36	import org.apache.carbondata.core.util.path.CarbonTablePath;
37	import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
38	import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
39	import org.apache.carbondata.processing.util.CarbonLoaderUtil;
40
41	import org.apache.commons.lang3.StringUtils;
42	import org.apache.log4j.Logger;
43
44	public class TableProcessingOperations {	×
45	private static final Logger LOGGER =	1✔
46	LogServiceFactory.getLogService(CarbonLoaderUtil.class.getName());	1✔
47
48	/**
49	*
50	* @param carbonTable
51	* @param isCompactionFlow
52	* @throws IOException
53	*/
54	public static void deletePartialLoadDataIfExist(CarbonTable carbonTable,
55	final boolean isCompactionFlow) throws IOException {
56	String metaDataLocation = carbonTable.getMetadataPath();	1✔
57	final LoadMetadataDetails[] details = SegmentStatusManager.readLoadMetadata(metaDataLocation);	1✔
58
59	//delete folder which metadata no exist in tablestatus
60	String partitionPath = CarbonTablePath.getPartitionDir(carbonTable.getTablePath());	1✔
61	FileFactory.FileType fileType = FileFactory.getFileType(partitionPath);	1✔
62	if (FileFactory.isFileExist(partitionPath, fileType)) {	1✔
63	CarbonFile carbonFile = FileFactory.getCarbonFile(partitionPath, fileType);	1✔
64	CarbonFile[] listFiles = carbonFile.listFiles(new CarbonFileFilter() {	1✔
65	@Override public boolean accept(CarbonFile path) {
66	String segmentId =	1✔
67	CarbonTablePath.DataFileUtil.getSegmentIdFromPath(path.getAbsolutePath() + "/dummy");	1✔
68	boolean found = false;	1✔
69	for (int j = 0; j < details.length; j++) {	1✔
70	if (details[j].getLoadName().equals(segmentId)) {	1✔
71	found = true;	1✔
72	break;	1✔
73	}
74	}
75	return !found;	1✔
76	}
77	});
78	for (int k = 0; k < listFiles.length; k++) {	1✔
79	String segmentId = CarbonTablePath.DataFileUtil	1✔
80	.getSegmentIdFromPath(listFiles[k].getAbsolutePath() + "/dummy");	1✔
81	if (isCompactionFlow) {	1✔
82	if (segmentId.contains(".")) {	×
83	CarbonLoaderUtil.deleteStorePath(listFiles[k].getAbsolutePath());	×
84	}
85	} else {
86	if (!segmentId.contains(".")) {	1✔
87	CarbonLoaderUtil.deleteStorePath(listFiles[k].getAbsolutePath());	1✔
88	}
89	}
90	}
91	}
92	}	1✔
93
94	/**
95	*
96	* This method will delete the local data load folder location after data load is complete
97	*
98	* @param loadModel
99	* @param isCompactionFlow COMPACTION keyword will be added to path to make path unique if true
100	* @param isAltPartitionFlow Alter_Partition keyword will be added to path to make path unique if
101	* true
102	*/
103	public static void deleteLocalDataLoadFolderLocation(CarbonLoadModel loadModel,
104	boolean isCompactionFlow, boolean isAltPartitionFlow) {
105	String tableName = loadModel.getTableName();	1✔
106	String databaseName = loadModel.getDatabaseName();	1✔
107	String tempLocationKey = CarbonDataProcessorUtil	1✔
108	.getTempStoreLocationKey(databaseName, tableName, loadModel.getSegmentId(),	1✔
109	loadModel.getTaskNo(), isCompactionFlow, isAltPartitionFlow);	1✔
110	deleteLocalDataLoadFolderLocation(tempLocationKey, tableName);	1✔
111	}	1✔
112
113	/**
114	*
115	* This method will delete the local data load folder location after data load is complete
116	*
117	* @param tempLocationKey temporary location set in carbon properties
118	* @param tableName
119	*/
120	public static void deleteLocalDataLoadFolderLocation(String tempLocationKey, String tableName) {
121
122	// form local store location
123	final String localStoreLocations = CarbonProperties.getInstance().getProperty(tempLocationKey);	1✔
124	if (localStoreLocations == null) {	1✔
125	throw new RuntimeException("Store location not set for the key " + tempLocationKey);	×
126	}
127	// submit local folder clean up in another thread so that main thread execution is not blocked
128	ExecutorService localFolderDeletionService = Executors	1✔
129	.newFixedThreadPool(1, new CarbonThreadFactory("LocalFolderDeletionPool:" + tableName));	1✔
130	try {
131	localFolderDeletionService.submit(new Callable<Void>() {	1✔
132	@Override public Void call() throws Exception {
133	long startTime = System.currentTimeMillis();	1✔
134	String[] locArray = StringUtils.split(localStoreLocations, File.pathSeparator);	1✔
135	for (String loc : locArray) {	1✔
136	try {
137	CarbonUtil.deleteFoldersAndFiles(new File(loc));	1✔
138	} catch (IOException \| InterruptedException e) {	×
139	LOGGER.error("Failed to delete local data load folder location: " + loc, e);	×
140	}	1✔
141	}
142	LOGGER.info(	1✔
143	"Deleted the local store location: " + localStoreLocations + " : Time taken: " + (
144	System.currentTimeMillis() - startTime));	1✔
145	return null;	1✔
146	}
147	});
148	} finally {
149	CarbonProperties.getInstance().removeProperty(tempLocationKey);	1✔
150	if (null != localFolderDeletionService) {	1✔
151	localFolderDeletionService.shutdown();	1✔
152	}
153	}	×
154
155	}	1✔
156	}

apache / carbondata / 1748

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