apache / carbondata / 797

Build # 797

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/3a1bb5a7151de28305a767627f6bc01a1270d03a">3a1bb5a71<a href="https://github.com/apache/carbondata/commit/3a1bb5a7151de28305a767627f6bc01a1270d03a">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/d635431b0891df4f7da8b4726f94a206f8d4e626">d635431b0</a>">d635431b0</a><a href="https://github.com/apache/carbondata/commit/3a1bb5a7151de28305a767627f6bc01a1270d03a"> into 5d17ff40b">5d17ff40b</a>

Run Details

18 of 18 new or added lines in 2 files covered. (100.0%)

60473 of 75045 relevant lines covered (80.58%)

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.*;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.carbondata.common.logging.LogService;
import org.apache.carbondata.common.logging.LogServiceFactory;
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;

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

apache / carbondata / 797

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