2583

Build # 2583

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/2e46382df39eec3798f4f1183ba5fdf5bcdeff03">2e46382df<a href="https://github.com/apache/carbondata/commit/2e46382df39eec3798f4f1183ba5fdf5bcdeff03">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/6745f4e2a245e36b313ed145507a7a40b212440b">6745f4e2a</a>">6745f4e2a</a><a href="https://github.com/apache/carbondata/commit/2e46382df39eec3798f4f1183ba5fdf5bcdeff03"> into 6605b54d5">6605b54d5</a>

Run Details

62486 of 78769 relevant lines covered (79.33%)

1.05 hits per line

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

85.54

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

apache / carbondata / 2583

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