core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlockletWrapper.java - carbondata - Git at Google

 /*
  * 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.indexstore;

 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.DataInput;
 import java.io.DataInputStream;
 import java.io.DataOutput;
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;

 import org.apache.carbondata.common.logging.LogServiceFactory;
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
 import org.apache.carbondata.core.datastore.compression.SnappyCompressor;
 import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
 import org.apache.carbondata.core.datastore.impl.FileFactory;
 import org.apache.carbondata.core.metadata.schema.table.Writable;
 import org.apache.carbondata.core.stream.ExtendedByteArrayInputStream;
 import org.apache.carbondata.core.stream.ExtendedByteArrayOutputStream;
 import org.apache.carbondata.core.stream.ExtendedDataInputStream;
 import org.apache.carbondata.core.util.CarbonProperties;
 import org.apache.carbondata.core.util.CarbonUtil;

 import org.apache.log4j.Logger;

 /**
  * class will be used to send extended blocklet object from index executor to index driver
  * if data size is more than it will be written in temp folder provided by user
  * and only file name will send, if data size is less then complete data will be send to
  * index executor
  */
 public class ExtendedBlockletWrapper implements Writable, Serializable {

   private static final Logger LOGGER =
       LogServiceFactory.getLogService(ExtendedBlockletWrapper.class.getName());
   private static final int BUFFER_SIZE = 8 * 1024 * 1024;
   private static final int BLOCK_SIZE = 256 * 1024 * 1024;
   private boolean isWrittenToFile;
   private int dataSize;
   private byte[] bytes;

   public ExtendedBlockletWrapper() {

   }

   public ExtendedBlockletWrapper(List<ExtendedBlocklet> extendedBlockletList, String tablePath,
       String queryId, boolean isWriteToFile, boolean isCountJob) {
     Map<String, Short> uniqueLocations = new HashMap<>();
     byte[] bytes = convertToBytes(tablePath, uniqueLocations, extendedBlockletList, isCountJob);
     int serializeAllowedSize = Integer.parseInt(CarbonProperties.getInstance()
         .getProperty(CarbonCommonConstants.CARBON_INDEX_SERVER_SERIALIZATION_THRESHOLD,
             CarbonCommonConstants.CARBON_INDEX_SERVER_SERIALIZATION_THRESHOLD_DEFAULT)) * 1024;
     DataOutputStream stream = null;
     // if data size is more then data will be written in file and file name will be sent from
     // executor to driver, in case of any failure data will send through network
     if (bytes.length > serializeAllowedSize && isWriteToFile) {
       final String fileName = UUID.randomUUID().toString();
       String folderPath = CarbonUtil.getIndexServerTempPath()
               + CarbonCommonConstants.FILE_SEPARATOR + queryId;
       try {
         final CarbonFile carbonFile = FileFactory.getCarbonFile(folderPath);
         boolean isFolderExists = true;
         if (!carbonFile.isFileExist()) {
           LOGGER.warn("Folder:" + folderPath + "doesn't exists, data will be send through network");
           isFolderExists = false;
         }
         if (isFolderExists) {
           stream = FileFactory.getDataOutputStream(folderPath + "/" + fileName,
                   BUFFER_SIZE, BLOCK_SIZE, (short) 1);
           writeBlockletToStream(stream, bytes, uniqueLocations, extendedBlockletList);
           this.dataSize = stream.size();
           this.bytes = fileName.getBytes(CarbonCommonConstants.DEFAULT_CHARSET);
           isWrittenToFile = true;
         }
       } catch (IOException e) {
         LOGGER.error("Problem while writing to file, data will be sent through network", e);
       } finally {
         CarbonUtil.closeStreams(stream);
       }
     }
     if (!isWrittenToFile) {
       try {
         ExtendedByteArrayOutputStream bos = new ExtendedByteArrayOutputStream();
         stream = new DataOutputStream(bos);
         writeBlockletToStream(stream, bytes, uniqueLocations, extendedBlockletList);
         this.dataSize = bos.size();
         this.bytes = bos.getBuffer();
       } catch (IOException e) {
         LOGGER.error("Problem while writing data to memory stream", e);
       } finally {
         CarbonUtil.closeStreams(stream);
       }
     }
   }

   private byte[] convertToBytes(String tablePath, Map<String, Short> uniqueLocations,
       List<ExtendedBlocklet> extendedBlockletList, boolean isCountJob) {
     ByteArrayOutputStream bos = new ExtendedByteArrayOutputStream();
     DataOutputStream stream = new DataOutputStream(bos);
     try {
       for (ExtendedBlocklet extendedBlocklet : extendedBlockletList) {
         extendedBlocklet.setFilePath(extendedBlocklet.getFilePath().replace(tablePath, ""));
         extendedBlocklet.serializeData(stream, uniqueLocations, isCountJob);
       }
       byte[] input = bos.toByteArray();
       return new SnappyCompressor().compressByte(input, input.length);
     } catch (IOException e) {
       throw new RuntimeException(e);
     } finally {
       CarbonUtil.closeStreams(stream);
     }
   }

   /**
    * Below method will be used to write the data to stream[file/memory]
    * Data Format
    * <number of splits><number of unique location[short]><locations><serialize data len><data>
    *
    * @param stream
    * @param data
    * @param uniqueLocation
    * @param extendedBlockletList
    * @throws IOException
    */
   private void writeBlockletToStream(DataOutputStream stream, byte[] data,
       Map<String, Short> uniqueLocation, List<ExtendedBlocklet> extendedBlockletList)
       throws IOException {
     stream.writeInt(extendedBlockletList.size());
     String[] uniqueLoc = new String[uniqueLocation.size()];
     Iterator<Map.Entry<String, Short>> iterator = uniqueLocation.entrySet().iterator();
     while (iterator.hasNext()) {
       final Map.Entry<String, Short> next = iterator.next();
       uniqueLoc[next.getValue()] = next.getKey();
     }
     stream.writeShort((short) uniqueLoc.length);
     for (String loc : uniqueLoc) {
       stream.writeUTF(loc);
     }
     stream.writeInt(data.length);
     stream.write(data);
   }

   /**
    * deserialize the blocklet data from file or stream
    * data format
    * <number of splits><number of unique location[short]><locations><serialize data len><data>
    *
    * @param tablePath
    * @param queryId
    * @return
    * @throws IOException
    */
   public List<ExtendedBlocklet> readBlocklet(String tablePath, String queryId, boolean isCountJob)
       throws IOException {
     byte[] data;
     if (bytes != null) {
       if (isWrittenToFile) {
         DataInputStream stream = null;
         try {
           final String folderPath = CarbonUtil.getIndexServerTempPath()
                   + CarbonCommonConstants.FILE_SEPARATOR + queryId;
           String fileName = new String(bytes, CarbonCommonConstants.DEFAULT_CHARSET);
           stream = FileFactory
               .getDataInputStream(folderPath + "/" + fileName);
           data = new byte[dataSize];
           stream.readFully(data);
         } finally {
           CarbonUtil.closeStreams(stream);
         }
       } else {
         data = bytes;
       }
       DataInputStream stream = null;
       int numberOfBlocklet;
       String[] locations;
       int actualDataLen;
       try {
         stream = new DataInputStream(new ByteArrayInputStream(data));
         numberOfBlocklet = stream.readInt();
         short numberOfLocations = stream.readShort();
         locations = new String[numberOfLocations];
         for (int i = 0; i < numberOfLocations; i++) {
           locations[i] = stream.readUTF();
         }
         actualDataLen = stream.readInt();
       } finally {
         CarbonUtil.closeStreams(stream);
       }

       final byte[] unCompressByte =
           new SnappyCompressor().unCompressByte(data, this.dataSize - actualDataLen, actualDataLen);
       ExtendedByteArrayInputStream ebis = new ExtendedByteArrayInputStream(unCompressByte);
       ExtendedDataInputStream eDIS = new ExtendedDataInputStream(ebis);
       List<ExtendedBlocklet> extendedBlockletList = new ArrayList<>();
       try {
         for (int i = 0; i < numberOfBlocklet; i++) {
           ExtendedBlocklet extendedBlocklet = new ExtendedBlocklet();
           extendedBlocklet.deserializeFields(eDIS, locations, tablePath, isCountJob);
           extendedBlockletList.add(extendedBlocklet);
         }
       } finally {
         CarbonUtil.closeStreams(eDIS);
       }
       return extendedBlockletList;
     } else {
       return new ArrayList<>();
     }
   }

   @Override
   public void write(DataOutput out) throws IOException {
     out.writeBoolean(isWrittenToFile);
     out.writeBoolean(bytes != null);
     if (bytes != null) {
       out.writeInt(bytes.length);
       out.write(bytes);
     }
     out.writeInt(dataSize);
   }

   @Override
   public void readFields(DataInput in) throws IOException {
     this.isWrittenToFile = in.readBoolean();
     if (in.readBoolean()) {
       this.bytes = new byte[in.readInt()];
       in.readFully(bytes);
     }
     this.dataSize = in.readInt();
   }

 }
	/*
	* 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.indexstore;

	import java.io.ByteArrayInputStream;
	import java.io.ByteArrayOutputStream;
	import java.io.DataInput;
	import java.io.DataInputStream;
	import java.io.DataOutput;
	import java.io.DataOutputStream;
	import java.io.IOException;
	import java.io.Serializable;
	import java.util.ArrayList;
	import java.util.HashMap;
	import java.util.Iterator;
	import java.util.List;
	import java.util.Map;
	import java.util.UUID;

	import org.apache.carbondata.common.logging.LogServiceFactory;
	import org.apache.carbondata.core.constants.CarbonCommonConstants;
	import org.apache.carbondata.core.datastore.compression.SnappyCompressor;
	import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
	import org.apache.carbondata.core.datastore.impl.FileFactory;
	import org.apache.carbondata.core.metadata.schema.table.Writable;
	import org.apache.carbondata.core.stream.ExtendedByteArrayInputStream;
	import org.apache.carbondata.core.stream.ExtendedByteArrayOutputStream;
	import org.apache.carbondata.core.stream.ExtendedDataInputStream;
	import org.apache.carbondata.core.util.CarbonProperties;
	import org.apache.carbondata.core.util.CarbonUtil;

	import org.apache.log4j.Logger;

	/**
	* class will be used to send extended blocklet object from index executor to index driver
	* if data size is more than it will be written in temp folder provided by user
	* and only file name will send, if data size is less then complete data will be send to
	* index executor
	*/
	public class ExtendedBlockletWrapper implements Writable, Serializable {

	private static final Logger LOGGER =
	LogServiceFactory.getLogService(ExtendedBlockletWrapper.class.getName());
	private static final int BUFFER_SIZE = 8 * 1024 * 1024;
	private static final int BLOCK_SIZE = 256 * 1024 * 1024;
	private boolean isWrittenToFile;
	private int dataSize;
	private byte[] bytes;

	public ExtendedBlockletWrapper() {

	}

	public ExtendedBlockletWrapper(List<ExtendedBlocklet> extendedBlockletList, String tablePath,
	String queryId, boolean isWriteToFile, boolean isCountJob) {
	Map<String, Short> uniqueLocations = new HashMap<>();
	byte[] bytes = convertToBytes(tablePath, uniqueLocations, extendedBlockletList, isCountJob);
	int serializeAllowedSize = Integer.parseInt(CarbonProperties.getInstance()
	.getProperty(CarbonCommonConstants.CARBON_INDEX_SERVER_SERIALIZATION_THRESHOLD,
	CarbonCommonConstants.CARBON_INDEX_SERVER_SERIALIZATION_THRESHOLD_DEFAULT)) * 1024;
	DataOutputStream stream = null;
	// if data size is more then data will be written in file and file name will be sent from
	// executor to driver, in case of any failure data will send through network
	if (bytes.length > serializeAllowedSize && isWriteToFile) {
	final String fileName = UUID.randomUUID().toString();
	String folderPath = CarbonUtil.getIndexServerTempPath()
	+ CarbonCommonConstants.FILE_SEPARATOR + queryId;
	try {
	final CarbonFile carbonFile = FileFactory.getCarbonFile(folderPath);
	boolean isFolderExists = true;
	if (!carbonFile.isFileExist()) {
	LOGGER.warn("Folder:" + folderPath + "doesn't exists, data will be send through network");
	isFolderExists = false;
	}
	if (isFolderExists) {
	stream = FileFactory.getDataOutputStream(folderPath + "/" + fileName,
	BUFFER_SIZE, BLOCK_SIZE, (short) 1);
	writeBlockletToStream(stream, bytes, uniqueLocations, extendedBlockletList);
	this.dataSize = stream.size();
	this.bytes = fileName.getBytes(CarbonCommonConstants.DEFAULT_CHARSET);
	isWrittenToFile = true;
	}
	} catch (IOException e) {
	LOGGER.error("Problem while writing to file, data will be sent through network", e);
	} finally {
	CarbonUtil.closeStreams(stream);
	}
	}
	if (!isWrittenToFile) {
	try {
	ExtendedByteArrayOutputStream bos = new ExtendedByteArrayOutputStream();
	stream = new DataOutputStream(bos);
	writeBlockletToStream(stream, bytes, uniqueLocations, extendedBlockletList);
	this.dataSize = bos.size();
	this.bytes = bos.getBuffer();
	} catch (IOException e) {
	LOGGER.error("Problem while writing data to memory stream", e);
	} finally {
	CarbonUtil.closeStreams(stream);
	}
	}
	}

	private byte[] convertToBytes(String tablePath, Map<String, Short> uniqueLocations,
	List<ExtendedBlocklet> extendedBlockletList, boolean isCountJob) {
	ByteArrayOutputStream bos = new ExtendedByteArrayOutputStream();
	DataOutputStream stream = new DataOutputStream(bos);
	try {
	for (ExtendedBlocklet extendedBlocklet : extendedBlockletList) {
	extendedBlocklet.setFilePath(extendedBlocklet.getFilePath().replace(tablePath, ""));
	extendedBlocklet.serializeData(stream, uniqueLocations, isCountJob);
	}
	byte[] input = bos.toByteArray();
	return new SnappyCompressor().compressByte(input, input.length);
	} catch (IOException e) {
	throw new RuntimeException(e);
	} finally {
	CarbonUtil.closeStreams(stream);
	}
	}

	/**
	* Below method will be used to write the data to stream[file/memory]
	* Data Format
	* <number of splits><number of unique location[short]><locations><serialize data len><data>
	*
	* @param stream
	* @param data
	* @param uniqueLocation
	* @param extendedBlockletList
	* @throws IOException
	*/
	private void writeBlockletToStream(DataOutputStream stream, byte[] data,
	Map<String, Short> uniqueLocation, List<ExtendedBlocklet> extendedBlockletList)
	throws IOException {
	stream.writeInt(extendedBlockletList.size());
	String[] uniqueLoc = new String[uniqueLocation.size()];
	Iterator<Map.Entry<String, Short>> iterator = uniqueLocation.entrySet().iterator();
	while (iterator.hasNext()) {
	final Map.Entry<String, Short> next = iterator.next();
	uniqueLoc[next.getValue()] = next.getKey();
	}
	stream.writeShort((short) uniqueLoc.length);
	for (String loc : uniqueLoc) {
	stream.writeUTF(loc);
	}
	stream.writeInt(data.length);
	stream.write(data);
	}

	/**
	* deserialize the blocklet data from file or stream
	* data format
	* <number of splits><number of unique location[short]><locations><serialize data len><data>
	*
	* @param tablePath
	* @param queryId
	* @return
	* @throws IOException
	*/
	public List<ExtendedBlocklet> readBlocklet(String tablePath, String queryId, boolean isCountJob)
	throws IOException {
	byte[] data;
	if (bytes != null) {
	if (isWrittenToFile) {
	DataInputStream stream = null;
	try {
	final String folderPath = CarbonUtil.getIndexServerTempPath()
	+ CarbonCommonConstants.FILE_SEPARATOR + queryId;
	String fileName = new String(bytes, CarbonCommonConstants.DEFAULT_CHARSET);
	stream = FileFactory
	.getDataInputStream(folderPath + "/" + fileName);
	data = new byte[dataSize];
	stream.readFully(data);
	} finally {
	CarbonUtil.closeStreams(stream);
	}
	} else {
	data = bytes;
	}
	DataInputStream stream = null;
	int numberOfBlocklet;
	String[] locations;
	int actualDataLen;
	try {
	stream = new DataInputStream(new ByteArrayInputStream(data));
	numberOfBlocklet = stream.readInt();
	short numberOfLocations = stream.readShort();
	locations = new String[numberOfLocations];
	for (int i = 0; i < numberOfLocations; i++) {
	locations[i] = stream.readUTF();
	}
	actualDataLen = stream.readInt();
	} finally {
	CarbonUtil.closeStreams(stream);
	}

	final byte[] unCompressByte =
	new SnappyCompressor().unCompressByte(data, this.dataSize - actualDataLen, actualDataLen);
	ExtendedByteArrayInputStream ebis = new ExtendedByteArrayInputStream(unCompressByte);
	ExtendedDataInputStream eDIS = new ExtendedDataInputStream(ebis);
	List<ExtendedBlocklet> extendedBlockletList = new ArrayList<>();
	try {
	for (int i = 0; i < numberOfBlocklet; i++) {
	ExtendedBlocklet extendedBlocklet = new ExtendedBlocklet();
	extendedBlocklet.deserializeFields(eDIS, locations, tablePath, isCountJob);
	extendedBlockletList.add(extendedBlocklet);
	}
	} finally {
	CarbonUtil.closeStreams(eDIS);
	}
	return extendedBlockletList;
	} else {
	return new ArrayList<>();
	}
	}

	@Override
	public void write(DataOutput out) throws IOException {
	out.writeBoolean(isWrittenToFile);
	out.writeBoolean(bytes != null);
	if (bytes != null) {
	out.writeInt(bytes.length);
	out.write(bytes);
	}
	out.writeInt(dataSize);
	}

	@Override
	public void readFields(DataInput in) throws IOException {
	this.isWrittenToFile = in.readBoolean();
	if (in.readBoolean()) {
	this.bytes = new byte[in.readInt()];
	in.readFully(bytes);
	}
	this.dataSize = in.readInt();
	}

	}