1376

Build # 1376

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/5398c94a021e90119462d3701bbe15eb81acdee3">5398c94a0<a href="https://github.com/apache/carbondata/commit/5398c94a021e90119462d3701bbe15eb81acdee3">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/6f718ba8ca5753a385cda9b3aa20921e07226cdd">6f718ba8c</a>">6f718ba8c</a><a href="https://github.com/apache/carbondata/commit/5398c94a021e90119462d3701bbe15eb81acdee3"> into d27f20523">d27f20523</a>

Run Details

14 of 14 new or added lines in 5 files covered. (100.0%)

61770 of 77134 relevant lines covered (80.08%)

1.06 hits per line

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

84.52

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

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.common.logging.impl.Audit;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.fileoperations.AtomicFileOperationFactory;
import org.apache.carbondata.core.fileoperations.AtomicFileOperations;
import org.apache.carbondata.core.fileoperations.FileWriteOperation;
import org.apache.carbondata.core.locks.CarbonLockFactory;
import org.apache.carbondata.core.locks.CarbonLockUtil;
import org.apache.carbondata.core.locks.ICarbonLock;
import org.apache.carbondata.core.locks.LockUsage;
import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.core.util.CarbonUtil;

import com.google.gson.Gson;
import org.apache.log4j.Logger;

/**
 * It saves/serializes the array of {{@link DataMapStatusDetail}} to disk in json format.
 * It ensures the data consistance while concurrent write through write lock. It saves the status
 * to the datamapstatus under the system folder.
 */
public class DiskBasedDataMapStatusProvider implements DataMapStatusStorageProvider {

  private static final Logger LOG =
      LogServiceFactory.getLogService(DiskBasedDataMapStatusProvider.class.getName());

  private static final String DATAMAP_STATUS_FILE = "datamapstatus";

  @Override
  public DataMapStatusDetail[] getDataMapStatusDetails() throws IOException {
    String statusPath = CarbonProperties.getInstance().getSystemFolderLocation()
        + CarbonCommonConstants.FILE_SEPARATOR + DATAMAP_STATUS_FILE;
    Gson gsonObjectToRead = new Gson();
    DataInputStream dataInputStream = null;
    BufferedReader buffReader = null;
    InputStreamReader inStream = null;
    DataMapStatusDetail[] dataMapStatusDetails;
    try {
      if (!FileFactory.isFileExist(statusPath)) {
        return new DataMapStatusDetail[0];
      }
      dataInputStream =
          FileFactory.getDataInputStream(statusPath, FileFactory.getFileType(statusPath));
      inStream = new InputStreamReader(dataInputStream,
          Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET));
      buffReader = new BufferedReader(inStream);
      dataMapStatusDetails = gsonObjectToRead.fromJson(buffReader, DataMapStatusDetail[].class);
    } catch (IOException e) {
      LOG.error("Failed to read datamap status", e);
      throw e;
    } finally {
      CarbonUtil.closeStreams(buffReader, inStream, dataInputStream);
    }

    // if dataMapStatusDetails is null, return empty array
    if (null == dataMapStatusDetails) {
      return new DataMapStatusDetail[0];
    }

    return dataMapStatusDetails;
  }

  /**
   * Update or add the status of passed datamaps with the given datamapstatus. If the datamapstatus
   * given is enabled/disabled then updates/adds the datamap, in case of drop it just removes it
   * from the file.
   * This method always overwrites the old file.
   * @param dataMapSchemas schemas of which are need to be updated in datamap status
   * @param dataMapStatus  status to be updated for the datamap schemas
   * @throws IOException
   */
  @Override
  public void updateDataMapStatus(List<DataMapSchema> dataMapSchemas, DataMapStatus dataMapStatus)
      throws IOException {
    if (dataMapSchemas == null || dataMapSchemas.size() == 0) {
      // There is nothing to update
      return;
    }
    ICarbonLock carbonTableStatusLock = getDataMapStatusLock();
    boolean locked = false;
    try {
      locked = carbonTableStatusLock.lockWithRetries();
      if (locked) {
        LOG.info("Datamap status lock has been successfully acquired.");
        DataMapStatusDetail[] dataMapStatusDetails = getDataMapStatusDetails();
        List<DataMapStatusDetail> dataMapStatusList = Arrays.asList(dataMapStatusDetails);
        dataMapStatusList = new ArrayList<>(dataMapStatusList);
        List<DataMapStatusDetail> changedStatusDetails = new ArrayList<>();
        List<DataMapStatusDetail> newStatusDetails = new ArrayList<>();
        for (DataMapSchema dataMapSchema : dataMapSchemas) {
          boolean exists = false;
          for (DataMapStatusDetail statusDetail : dataMapStatusList) {
            if (statusDetail.getDataMapName().equals(dataMapSchema.getDataMapName())) {
              statusDetail.setStatus(dataMapStatus);
              changedStatusDetails.add(statusDetail);
              exists = true;
            }
          }
          if (!exists) {
            newStatusDetails
                .add(new DataMapStatusDetail(dataMapSchema.getDataMapName(), dataMapStatus));
          }
        }
        // Add the newly added datamaps to the list.
        if (newStatusDetails.size() > 0 && dataMapStatus != DataMapStatus.DROPPED) {
          dataMapStatusList.addAll(newStatusDetails);
        }
        // In case of dropped datamap, just remove from the list.
        if (dataMapStatus == DataMapStatus.DROPPED) {
          dataMapStatusList.removeAll(changedStatusDetails);
        }
        writeLoadDetailsIntoFile(CarbonProperties.getInstance().getSystemFolderLocation()
                + CarbonCommonConstants.FILE_SEPARATOR + DATAMAP_STATUS_FILE,
            dataMapStatusList.toArray(new DataMapStatusDetail[dataMapStatusList.size()]));
      } else {
        String errorMsg = "Upadating datamapstatus is failed due to another process taken the lock"
            + " for updating it";
        Audit.log(LOG, errorMsg);
        LOG.error(errorMsg);
        throw new IOException(errorMsg + " Please try after some time.");
      }
    } finally {
      if (locked) {
        CarbonLockUtil.fileUnlock(carbonTableStatusLock, LockUsage.DATAMAP_STATUS_LOCK);
      }
    }
  }

  /**
   * writes datamap status details
   *
   * @param dataMapStatusDetails
   * @throws IOException
   */
  private static void writeLoadDetailsIntoFile(String location,
      DataMapStatusDetail[] dataMapStatusDetails) throws IOException {
    AtomicFileOperations fileWrite = AtomicFileOperationFactory.getAtomicFileOperations(location);
    BufferedWriter brWriter = null;
    DataOutputStream dataOutputStream = null;
    Gson gsonObjectToWrite = new Gson();
    // write the updated data into the datamap status file.
    try {
      dataOutputStream = fileWrite.openForWrite(FileWriteOperation.OVERWRITE);
      brWriter = new BufferedWriter(new OutputStreamWriter(dataOutputStream,
          Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));

      String metadataInstance = gsonObjectToWrite.toJson(dataMapStatusDetails);
      brWriter.write(metadataInstance);
    } catch (IOException ioe) {
      LOG.error("Error message: " + ioe.getLocalizedMessage());
      fileWrite.setFailed();
      throw ioe;
    } finally {
      if (null != brWriter) {
        brWriter.flush();
      }
      CarbonUtil.closeStreams(brWriter);
      fileWrite.close();
    }

  }

  private static ICarbonLock getDataMapStatusLock() {
    return CarbonLockFactory
        .getSystemLevelCarbonLockObj(CarbonProperties.getInstance().getSystemFolderLocation(),
            LockUsage.DATAMAP_STATUS_LOCK);
  }
}

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.status;
19
20	import java.io.BufferedReader;
21	import java.io.BufferedWriter;
22	import java.io.DataInputStream;
23	import java.io.DataOutputStream;
24	import java.io.IOException;
25	import java.io.InputStreamReader;
26	import java.io.OutputStreamWriter;
27	import java.nio.charset.Charset;
28	import java.util.ArrayList;
29	import java.util.Arrays;
30	import java.util.List;
31
32	import org.apache.carbondata.common.logging.LogServiceFactory;
33	import org.apache.carbondata.common.logging.impl.Audit;
34	import org.apache.carbondata.core.constants.CarbonCommonConstants;
35	import org.apache.carbondata.core.datastore.impl.FileFactory;
36	import org.apache.carbondata.core.fileoperations.AtomicFileOperationFactory;
37	import org.apache.carbondata.core.fileoperations.AtomicFileOperations;
38	import org.apache.carbondata.core.fileoperations.FileWriteOperation;
39	import org.apache.carbondata.core.locks.CarbonLockFactory;
40	import org.apache.carbondata.core.locks.CarbonLockUtil;
41	import org.apache.carbondata.core.locks.ICarbonLock;
42	import org.apache.carbondata.core.locks.LockUsage;
43	import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
44	import org.apache.carbondata.core.util.CarbonProperties;
45	import org.apache.carbondata.core.util.CarbonUtil;
46
47	import com.google.gson.Gson;
48	import org.apache.log4j.Logger;
49
50	/**
51	* It saves/serializes the array of {{@link DataMapStatusDetail}} to disk in json format.
52	* It ensures the data consistance while concurrent write through write lock. It saves the status
53	* to the datamapstatus under the system folder.
54	*/
55	public class DiskBasedDataMapStatusProvider implements DataMapStatusStorageProvider {	1✔
56
57	private static final Logger LOG =	1✔
58	LogServiceFactory.getLogService(DiskBasedDataMapStatusProvider.class.getName());	1✔
59
60	private static final String DATAMAP_STATUS_FILE = "datamapstatus";
61
62	@Override
63	public DataMapStatusDetail[] getDataMapStatusDetails() throws IOException {
64	String statusPath = CarbonProperties.getInstance().getSystemFolderLocation()	1✔
65	+ CarbonCommonConstants.FILE_SEPARATOR + DATAMAP_STATUS_FILE;
66	Gson gsonObjectToRead = new Gson();	1✔
67	DataInputStream dataInputStream = null;	1✔
68	BufferedReader buffReader = null;	1✔
69	InputStreamReader inStream = null;	1✔
70	DataMapStatusDetail[] dataMapStatusDetails;
71	try {
72	if (!FileFactory.isFileExist(statusPath)) {	1✔
73	return new DataMapStatusDetail[0];	1✔
74	}
75	dataInputStream =	1✔
76	FileFactory.getDataInputStream(statusPath, FileFactory.getFileType(statusPath));	1✔
77	inStream = new InputStreamReader(dataInputStream,	1✔
78	Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET));	1✔
79	buffReader = new BufferedReader(inStream);	1✔
80	dataMapStatusDetails = gsonObjectToRead.fromJson(buffReader, DataMapStatusDetail[].class);	1✔
81	} catch (IOException e) {	×
82	LOG.error("Failed to read datamap status", e);	×
83	throw e;	×
84	} finally {
85	CarbonUtil.closeStreams(buffReader, inStream, dataInputStream);	1✔
86	}	1✔
87
88	// if dataMapStatusDetails is null, return empty array
89	if (null == dataMapStatusDetails) {	1✔
90	return new DataMapStatusDetail[0];	×
91	}
92
93	return dataMapStatusDetails;	1✔
94	}
95
96	/**
97	* Update or add the status of passed datamaps with the given datamapstatus. If the datamapstatus
98	* given is enabled/disabled then updates/adds the datamap, in case of drop it just removes it
99	* from the file.
100	* This method always overwrites the old file.
101	* @param dataMapSchemas schemas of which are need to be updated in datamap status
102	* @param dataMapStatus status to be updated for the datamap schemas
103	* @throws IOException
104	*/
105	@Override
106	public void updateDataMapStatus(List<DataMapSchema> dataMapSchemas, DataMapStatus dataMapStatus)
107	throws IOException {
108	if (dataMapSchemas == null \|\| dataMapSchemas.size() == 0) {	1✔
109	// There is nothing to update
110	return;	1✔
111	}
112	ICarbonLock carbonTableStatusLock = getDataMapStatusLock();	1✔
113	boolean locked = false;	1✔
114	try {
115	locked = carbonTableStatusLock.lockWithRetries();	1✔
116	if (locked) {	1✔
117	LOG.info("Datamap status lock has been successfully acquired.");	1✔
118	DataMapStatusDetail[] dataMapStatusDetails = getDataMapStatusDetails();	1✔
119	List<DataMapStatusDetail> dataMapStatusList = Arrays.asList(dataMapStatusDetails);	1✔
120	dataMapStatusList = new ArrayList<>(dataMapStatusList);	1✔
121	List<DataMapStatusDetail> changedStatusDetails = new ArrayList<>();	1✔
122	List<DataMapStatusDetail> newStatusDetails = new ArrayList<>();	1✔
123	for (DataMapSchema dataMapSchema : dataMapSchemas) {	1✔
124	boolean exists = false;	1✔
125	for (DataMapStatusDetail statusDetail : dataMapStatusList) {	1✔
126	if (statusDetail.getDataMapName().equals(dataMapSchema.getDataMapName())) {	1✔
127	statusDetail.setStatus(dataMapStatus);	1✔
128	changedStatusDetails.add(statusDetail);	1✔
129	exists = true;	1✔
130	}
131	}	1✔
132	if (!exists) {	1✔
133	newStatusDetails	1✔
134	.add(new DataMapStatusDetail(dataMapSchema.getDataMapName(), dataMapStatus));	1✔
135	}
136	}	1✔
137	// Add the newly added datamaps to the list.
138	if (newStatusDetails.size() > 0 && dataMapStatus != DataMapStatus.DROPPED) {	1✔
139	dataMapStatusList.addAll(newStatusDetails);	1✔
140	}
141	// In case of dropped datamap, just remove from the list.
142	if (dataMapStatus == DataMapStatus.DROPPED) {	1✔
143	dataMapStatusList.removeAll(changedStatusDetails);	1✔
144	}
145	writeLoadDetailsIntoFile(CarbonProperties.getInstance().getSystemFolderLocation()	1✔
146	+ CarbonCommonConstants.FILE_SEPARATOR + DATAMAP_STATUS_FILE,
147	dataMapStatusList.toArray(new DataMapStatusDetail[dataMapStatusList.size()]));	1✔
148	} else {	1✔
149	String errorMsg = "Upadating datamapstatus is failed due to another process taken the lock"	×
150	+ " for updating it";
151	Audit.log(LOG, errorMsg);	×
152	LOG.error(errorMsg);	×
153	throw new IOException(errorMsg + " Please try after some time.");	×
154	}
155	} finally {
156	if (locked) {	1✔
157	CarbonLockUtil.fileUnlock(carbonTableStatusLock, LockUsage.DATAMAP_STATUS_LOCK);	1✔
158	}
159	}	×
160	}	1✔
161
162	/**
163	* writes datamap status details
164	*
165	* @param dataMapStatusDetails
166	* @throws IOException
167	*/
168	private static void writeLoadDetailsIntoFile(String location,
169	DataMapStatusDetail[] dataMapStatusDetails) throws IOException {
170	AtomicFileOperations fileWrite = AtomicFileOperationFactory.getAtomicFileOperations(location);	1✔
171	BufferedWriter brWriter = null;	1✔
172	DataOutputStream dataOutputStream = null;	1✔
173	Gson gsonObjectToWrite = new Gson();	1✔
174	// write the updated data into the datamap status file.
175	try {
176	dataOutputStream = fileWrite.openForWrite(FileWriteOperation.OVERWRITE);	1✔
177	brWriter = new BufferedWriter(new OutputStreamWriter(dataOutputStream,	1✔
178	Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));	1✔
179
180	String metadataInstance = gsonObjectToWrite.toJson(dataMapStatusDetails);	1✔
181	brWriter.write(metadataInstance);	1✔
182	} catch (IOException ioe) {	×
183	LOG.error("Error message: " + ioe.getLocalizedMessage());	×
184	fileWrite.setFailed();	×
185	throw ioe;	×
186	} finally {
187	if (null != brWriter) {	1✔
188	brWriter.flush();	1✔
189	}
190	CarbonUtil.closeStreams(brWriter);	1✔
191	fileWrite.close();	1✔
192	}	1✔
193
194	}	1✔
195
196	private static ICarbonLock getDataMapStatusLock() {
197	return CarbonLockFactory	1✔
198	.getSystemLevelCarbonLockObj(CarbonProperties.getInstance().getSystemFolderLocation(),	1✔
199	LockUsage.DATAMAP_STATUS_LOCK);
200	}
201	}

apache / carbondata / 1376

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