server/src/main/java/org/apache/iotdb/db/metadata/logfile/MLogTxtWriter.java - iotdb - 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.iotdb.db.metadata.logfile;

 import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
 import org.apache.iotdb.db.metadata.MetadataConstant;
 import org.apache.iotdb.db.metadata.MetadataOperationType;
 import org.apache.iotdb.db.qp.physical.crud.CreateTemplatePlan;
 import org.apache.iotdb.db.qp.physical.crud.SetSchemaTemplatePlan;
 import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
 import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
 import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
 import org.apache.iotdb.db.qp.physical.sys.DropContinuousQueryPlan;
 import org.apache.iotdb.db.qp.physical.sys.MNodePlan;
 import org.apache.iotdb.db.qp.physical.sys.MeasurementMNodePlan;
 import org.apache.iotdb.db.qp.physical.sys.SetUsingSchemaTemplatePlan;
 import org.apache.iotdb.db.qp.physical.sys.StorageGroupMNodePlan;
 import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
 import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
 import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
 import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
 import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import java.io.BufferedReader;
 import java.io.BufferedWriter;
 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.FileReader;
 import java.io.FileWriter;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;

 public class MLogTxtWriter implements AutoCloseable {

   private static final Logger logger = LoggerFactory.getLogger(MLogTxtWriter.class);
   private static final String STRING_TYPE = "%s,%s,%s" + System.lineSeparator();
   private static final String LINE_SEPARATOR = System.lineSeparator();
   private final File logFile;
   private FileOutputStream fileOutputStream;
   private FileChannel channel;
   private final AtomicInteger lineNumber;

   public MLogTxtWriter(String schemaDir, String logFileName) throws IOException {
     File metadataDir = SystemFileFactory.INSTANCE.getFile(schemaDir);
     if (!metadataDir.exists()) {
       if (metadataDir.mkdirs()) {
         logger.info("create schema folder {}.", metadataDir);
       } else {
         logger.info("create schema folder {} failed.", metadataDir);
       }
     }

     logFile = SystemFileFactory.INSTANCE.getFile(schemaDir + File.separator + logFileName);
     fileOutputStream = new FileOutputStream(logFile, true);
     channel = fileOutputStream.getChannel();
     lineNumber = new AtomicInteger(0);
   }

   public MLogTxtWriter(String logFileName) throws FileNotFoundException {
     logFile = SystemFileFactory.INSTANCE.getFile(logFileName);
     fileOutputStream = new FileOutputStream(logFile, true);
     channel = fileOutputStream.getChannel();
     lineNumber = new AtomicInteger(0);
   }

   @Override
   public void close() throws IOException {
     fileOutputStream.close();
   }

   public void createTimeseries(CreateTimeSeriesPlan plan, long offset) throws IOException {
     StringBuilder buf = new StringBuilder();
     buf.append(
         String.format(
             "%s,%s,%s,%s,%s",
             MetadataOperationType.CREATE_TIMESERIES,
             plan.getPath().getFullPath(),
             plan.getDataType().serialize(),
             plan.getEncoding().serialize(),
             plan.getCompressor().serialize()));

     buf.append(",");
     if (plan.getProps() != null) {
       boolean first = true;
       for (Map.Entry<String, String> entry : plan.getProps().entrySet()) {
         if (first) {
           buf.append(String.format("%s=%s", entry.getKey(), entry.getValue()));
           first = false;
         } else {
           buf.append(String.format("&%s=%s", entry.getKey(), entry.getValue()));
         }
       }
     }

     buf.append(",");
     if (plan.getAlias() != null) {
       buf.append(plan.getAlias());
     }

     buf.append(",");
     if (offset >= 0) {
       buf.append(offset);
     }
     buf.append(LINE_SEPARATOR);
     channel.write(ByteBuffer.wrap(buf.toString().getBytes()));
     lineNumber.incrementAndGet();
   }

   public void createAlignedTimeseries(CreateAlignedTimeSeriesPlan plan) throws IOException {
     StringBuilder buf = new StringBuilder();
     buf.append(
         String.format(
             "%s,%s,%s,%s,%s,%s",
             MetadataOperationType.CREATE_TIMESERIES,
             plan.getPrefixPath().getFullPath(),
             plan.getMeasurements(),
             plan.getDataTypes().stream().map(TSDataType::serialize),
             plan.getEncodings().stream().map(TSEncoding::serialize),
             plan.getCompressor().serialize()));

     buf.append(",[");
     if (plan.getAliasList() != null) {
       List<String> aliasList = plan.getAliasList();
       for (int i = 0; i < aliasList.size(); i++) {
         buf.append(aliasList.get(i));
         if (i != aliasList.size() - 1) {
           buf.append(",");
         }
       }
     }
     buf.append("]");
     buf.append(LINE_SEPARATOR);
     channel.write(ByteBuffer.wrap(buf.toString().getBytes()));
     lineNumber.incrementAndGet();
   }

   public void deleteTimeseries(String path) throws IOException {
     String outputStr = MetadataOperationType.DELETE_TIMESERIES + "," + path + LINE_SEPARATOR;
     ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
     channel.write(buff);
   }

   public void createContinuousQuery(CreateContinuousQueryPlan plan) throws IOException {
     String buf =
         String.format(
                 "%s,%s,%s,%s",
                 MetadataOperationType.CREATE_CONTINUOUS_QUERY,
                 plan.getContinuousQueryName(),
                 plan.getQuerySql(),
                 plan.getTargetPath().getFullPath())
             + LINE_SEPARATOR;
     channel.write(ByteBuffer.wrap(buf.getBytes()));
     lineNumber.incrementAndGet();
   }

   public void dropContinuousQuery(DropContinuousQueryPlan plan) throws IOException {

     String buf =
         String.format(
                 "%s,%s", MetadataOperationType.DROP_CONTINUOUS_QUERY, plan.getContinuousQueryName())
             + LINE_SEPARATOR;
     channel.write(ByteBuffer.wrap(buf.getBytes()));
     lineNumber.incrementAndGet();
   }

   public void setStorageGroup(String storageGroup) throws IOException {
     String outputStr =
         MetadataOperationType.SET_STORAGE_GROUP + "," + storageGroup + LINE_SEPARATOR;
     ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
     channel.write(buff);
     lineNumber.incrementAndGet();
   }

   public void deleteStorageGroup(String storageGroup) throws IOException {
     String outputStr =
         MetadataOperationType.DELETE_STORAGE_GROUP + "," + storageGroup + LINE_SEPARATOR;
     ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
     channel.write(buff);
     lineNumber.incrementAndGet();
   }

   public void setTTL(String storageGroup, long ttl) throws IOException {
     String outputStr = String.format(STRING_TYPE, MetadataOperationType.SET_TTL, storageGroup, ttl);
     ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
     channel.write(buff);
     lineNumber.incrementAndGet();
   }

   public void changeOffset(String path, long offset) throws IOException {
     String outputStr =
         String.format(STRING_TYPE, MetadataOperationType.CHANGE_OFFSET, path, offset);
     ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
     channel.write(buff);
     lineNumber.incrementAndGet();
   }

   public void changeAlias(String path, String alias) throws IOException {
     String outputStr = String.format(STRING_TYPE, MetadataOperationType.CHANGE_ALIAS, path, alias);
     ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
     channel.write(buff);
     lineNumber.incrementAndGet();
   }

   public static void upgradeMLog(String schemaDir, String logFileName) throws IOException {
     File logFile = SystemFileFactory.INSTANCE.getFile(schemaDir + File.separator + logFileName);
     File tmpLogFile = SystemFileFactory.INSTANCE.getFile(logFile.getAbsolutePath() + ".tmp");

     // if both old mlog and mlog.tmp do not exist, nothing to do
     if (!logFile.exists() && !tmpLogFile.exists()) {
       return;
     } else if (!logFile.exists() && tmpLogFile.exists()) {
       // if old mlog doesn't exsit but mlog.tmp exists, rename tmp file to mlog
       FSFactoryProducer.getFSFactory().moveFile(tmpLogFile, logFile);
       return;
     }

     // if both old mlog and mlog.tmp exist, delete mlog tmp, then do upgrading
     if (tmpLogFile.exists()) {
       try {
         Files.delete(Paths.get(tmpLogFile.toURI()));
       } catch (IOException e) {
         throw new IOException("Deleting " + tmpLogFile + "failed with exception " + e.getMessage());
       }
     }
     // upgrading
     try (BufferedReader reader = new BufferedReader(new FileReader(logFile));
         BufferedWriter writer = new BufferedWriter(new FileWriter(tmpLogFile, true))) {
       String line;
       while ((line = reader.readLine()) != null) {
         StringBuilder buf = new StringBuilder();
         buf.append(line);
         if (line.startsWith(MetadataOperationType.CREATE_TIMESERIES)) {
           buf.append(",,,");
         }
         writer.write(buf.toString());
         writer.newLine();
         writer.flush();
       }
     }
   }

   public void clear() throws IOException {
     channel.force(true);
     channel.close();
     fileOutputStream.close();
     Files.delete(logFile.toPath());
     fileOutputStream = new FileOutputStream(logFile, true);
     channel = fileOutputStream.getChannel();
     lineNumber.set(0);
   }

   public void serializeMNode(MNodePlan plan) throws IOException {
     StringBuilder s = new StringBuilder(String.valueOf(MetadataConstant.STORAGE_GROUP_MNODE_TYPE));
     s.append(",").append(plan.getName()).append(",");
     s.append(plan.getChildSize());
     s.append(LINE_SEPARATOR);
     ByteBuffer buff = ByteBuffer.wrap(s.toString().getBytes());
     channel.write(buff);
     lineNumber.incrementAndGet();
   }

   public void serializeMeasurementMNode(MeasurementMNodePlan plan) throws IOException {
     StringBuilder s = new StringBuilder(String.valueOf(MetadataConstant.MEASUREMENT_MNODE_TYPE));
     s.append(",").append(plan.getName()).append(",");
     if (plan.getAlias() != null) {
       s.append(plan.getAlias());
     }
     IMeasurementSchema schema = plan.getSchema();
     s.append(",").append(schema.getType().ordinal()).append(",");
     s.append(schema.getEncodingType().ordinal()).append(",");
     s.append(schema.getCompressor().ordinal()).append(",");
     if (schema.getProps() != null) {
       for (Map.Entry<String, String> entry : schema.getProps().entrySet()) {
         s.append(entry.getKey()).append(":").append(entry.getValue()).append(";");
       }
     }
     s.append(",").append(plan.getOffset()).append(",");
     s.append(plan.getChildSize());
     s.append(LINE_SEPARATOR);
     ByteBuffer buff = ByteBuffer.wrap(s.toString().getBytes());
     channel.write(buff);
     lineNumber.incrementAndGet();
   }

   public void serializeStorageGroupMNode(StorageGroupMNodePlan plan) throws IOException {
     StringBuilder s = new StringBuilder(String.valueOf(MetadataConstant.STORAGE_GROUP_MNODE_TYPE));
     s.append(",").append(plan.getName()).append(",");
     s.append(plan.getDataTTL()).append(",");
     s.append(plan.getChildSize());
     s.append(LINE_SEPARATOR);
     ByteBuffer buff = ByteBuffer.wrap(s.toString().getBytes());
     channel.write(buff);
     lineNumber.incrementAndGet();
   }

   public void setTemplate(SetSchemaTemplatePlan plan) throws IOException {
     StringBuilder buf = new StringBuilder(String.valueOf(MetadataOperationType.SET_TEMPLATE));
     buf.append(",");
     buf.append(plan.getTemplateName());
     buf.append(",");
     buf.append(plan.getPrefixPath());
     buf.append(LINE_SEPARATOR);
     ByteBuffer buff = ByteBuffer.wrap(buf.toString().getBytes());
     channel.write(buff);
     lineNumber.incrementAndGet();
   }

   public void setUsingTemplate(SetUsingSchemaTemplatePlan plan) throws IOException {
     StringBuilder buf = new StringBuilder(String.valueOf(MetadataOperationType.SET_USING_TEMPLATE));
     buf.append(",");
     buf.append(plan.getPrefixPath());
     buf.append(LINE_SEPARATOR);
     ByteBuffer buff = ByteBuffer.wrap(buf.toString().getBytes());
     channel.write(buff);
     lineNumber.incrementAndGet();
   }

   public void createTemplate(CreateTemplatePlan plan) throws IOException {
     StringBuilder buf = new StringBuilder();
     for (int i = 0; i < plan.getSchemaNames().size(); i++) {
       for (int j = 0; j < plan.getMeasurements().get(i).size(); j++) {
         String measurement;
         boolean isAligned = false;
         if (plan.getMeasurements().get(i).size() == 1) {
           measurement = plan.getSchemaNames().get(i);
         } else {
           // for aligned timeseries
           isAligned = true;
           measurement =
               plan.getSchemaNames().get(i)
                   + TsFileConstant.PATH_SEPARATOR
                   + plan.getMeasurements().get(i).get(j);
         }
         buf.append(
             String.format(
                 "%s,%s,%s,%s,%s,%s,%s",
                 MetadataOperationType.CREATE_TEMPLATE,
                 plan.getName(),
                 isAligned ? 1 : 0,
                 measurement,
                 plan.getDataTypes().get(i).get(j).serialize(),
                 plan.getEncodings().get(i).get(j).serialize(),
                 plan.getCompressors().get(i).serialize()));
         buf.append(LINE_SEPARATOR);
         lineNumber.incrementAndGet();
       }
     }
     ByteBuffer buff = ByteBuffer.wrap(buf.toString().getBytes());
     channel.write(buff);
   }

   public void autoCreateDeviceNode(String Device) throws IOException {
     String outputStr = MetadataOperationType.AUTO_CREATE_DEVICE + "," + Device + LINE_SEPARATOR;
     ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
     channel.write(buff);
     lineNumber.incrementAndGet();
   }

   int getLineNumber() {
     return lineNumber.get();
   }

   /** only used for initialize a mlog file writer. */
   void setLineNumber(int number) {
     lineNumber.set(number);
   }
 }
	/*
	* 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.iotdb.db.metadata.logfile;

	import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
	import org.apache.iotdb.db.metadata.MetadataConstant;
	import org.apache.iotdb.db.metadata.MetadataOperationType;
	import org.apache.iotdb.db.qp.physical.crud.CreateTemplatePlan;
	import org.apache.iotdb.db.qp.physical.crud.SetSchemaTemplatePlan;
	import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
	import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
	import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
	import org.apache.iotdb.db.qp.physical.sys.DropContinuousQueryPlan;
	import org.apache.iotdb.db.qp.physical.sys.MNodePlan;
	import org.apache.iotdb.db.qp.physical.sys.MeasurementMNodePlan;
	import org.apache.iotdb.db.qp.physical.sys.SetUsingSchemaTemplatePlan;
	import org.apache.iotdb.db.qp.physical.sys.StorageGroupMNodePlan;
	import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
	import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
	import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
	import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
	import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;

	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import java.io.BufferedReader;
	import java.io.BufferedWriter;
	import java.io.File;
	import java.io.FileNotFoundException;
	import java.io.FileOutputStream;
	import java.io.FileReader;
	import java.io.FileWriter;
	import java.io.IOException;
	import java.nio.ByteBuffer;
	import java.nio.channels.FileChannel;
	import java.nio.file.Files;
	import java.nio.file.Paths;
	import java.util.List;
	import java.util.Map;
	import java.util.concurrent.atomic.AtomicInteger;

	public class MLogTxtWriter implements AutoCloseable {

	private static final Logger logger = LoggerFactory.getLogger(MLogTxtWriter.class);
	private static final String STRING_TYPE = "%s,%s,%s" + System.lineSeparator();
	private static final String LINE_SEPARATOR = System.lineSeparator();
	private final File logFile;
	private FileOutputStream fileOutputStream;
	private FileChannel channel;
	private final AtomicInteger lineNumber;

	public MLogTxtWriter(String schemaDir, String logFileName) throws IOException {
	File metadataDir = SystemFileFactory.INSTANCE.getFile(schemaDir);
	if (!metadataDir.exists()) {
	if (metadataDir.mkdirs()) {
	logger.info("create schema folder {}.", metadataDir);
	} else {
	logger.info("create schema folder {} failed.", metadataDir);
	}
	}

	logFile = SystemFileFactory.INSTANCE.getFile(schemaDir + File.separator + logFileName);
	fileOutputStream = new FileOutputStream(logFile, true);
	channel = fileOutputStream.getChannel();
	lineNumber = new AtomicInteger(0);
	}

	public MLogTxtWriter(String logFileName) throws FileNotFoundException {
	logFile = SystemFileFactory.INSTANCE.getFile(logFileName);
	fileOutputStream = new FileOutputStream(logFile, true);
	channel = fileOutputStream.getChannel();
	lineNumber = new AtomicInteger(0);
	}

	@Override
	public void close() throws IOException {
	fileOutputStream.close();
	}

	public void createTimeseries(CreateTimeSeriesPlan plan, long offset) throws IOException {
	StringBuilder buf = new StringBuilder();
	buf.append(
	String.format(
	"%s,%s,%s,%s,%s",
	MetadataOperationType.CREATE_TIMESERIES,
	plan.getPath().getFullPath(),
	plan.getDataType().serialize(),
	plan.getEncoding().serialize(),
	plan.getCompressor().serialize()));

	buf.append(",");
	if (plan.getProps() != null) {
	boolean first = true;
	for (Map.Entry<String, String> entry : plan.getProps().entrySet()) {
	if (first) {
	buf.append(String.format("%s=%s", entry.getKey(), entry.getValue()));
	first = false;
	} else {
	buf.append(String.format("&%s=%s", entry.getKey(), entry.getValue()));
	}
	}
	}

	buf.append(",");
	if (plan.getAlias() != null) {
	buf.append(plan.getAlias());
	}

	buf.append(",");
	if (offset >= 0) {
	buf.append(offset);
	}
	buf.append(LINE_SEPARATOR);
	channel.write(ByteBuffer.wrap(buf.toString().getBytes()));
	lineNumber.incrementAndGet();
	}

	public void createAlignedTimeseries(CreateAlignedTimeSeriesPlan plan) throws IOException {
	StringBuilder buf = new StringBuilder();
	buf.append(
	String.format(
	"%s,%s,%s,%s,%s,%s",
	MetadataOperationType.CREATE_TIMESERIES,
	plan.getPrefixPath().getFullPath(),
	plan.getMeasurements(),
	plan.getDataTypes().stream().map(TSDataType::serialize),
	plan.getEncodings().stream().map(TSEncoding::serialize),
	plan.getCompressor().serialize()));

	buf.append(",[");
	if (plan.getAliasList() != null) {
	List<String> aliasList = plan.getAliasList();
	for (int i = 0; i < aliasList.size(); i++) {
	buf.append(aliasList.get(i));
	if (i != aliasList.size() - 1) {
	buf.append(",");
	}
	}
	}
	buf.append("]");
	buf.append(LINE_SEPARATOR);
	channel.write(ByteBuffer.wrap(buf.toString().getBytes()));
	lineNumber.incrementAndGet();
	}

	public void deleteTimeseries(String path) throws IOException {
	String outputStr = MetadataOperationType.DELETE_TIMESERIES + "," + path + LINE_SEPARATOR;
	ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
	channel.write(buff);
	}

	public void createContinuousQuery(CreateContinuousQueryPlan plan) throws IOException {
	String buf =
	String.format(
	"%s,%s,%s,%s",
	MetadataOperationType.CREATE_CONTINUOUS_QUERY,
	plan.getContinuousQueryName(),
	plan.getQuerySql(),
	plan.getTargetPath().getFullPath())
	+ LINE_SEPARATOR;
	channel.write(ByteBuffer.wrap(buf.getBytes()));
	lineNumber.incrementAndGet();
	}

	public void dropContinuousQuery(DropContinuousQueryPlan plan) throws IOException {

	String buf =
	String.format(
	"%s,%s", MetadataOperationType.DROP_CONTINUOUS_QUERY, plan.getContinuousQueryName())
	+ LINE_SEPARATOR;
	channel.write(ByteBuffer.wrap(buf.getBytes()));
	lineNumber.incrementAndGet();
	}

	public void setStorageGroup(String storageGroup) throws IOException {
	String outputStr =
	MetadataOperationType.SET_STORAGE_GROUP + "," + storageGroup + LINE_SEPARATOR;
	ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
	channel.write(buff);
	lineNumber.incrementAndGet();
	}

	public void deleteStorageGroup(String storageGroup) throws IOException {
	String outputStr =
	MetadataOperationType.DELETE_STORAGE_GROUP + "," + storageGroup + LINE_SEPARATOR;
	ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
	channel.write(buff);
	lineNumber.incrementAndGet();
	}

	public void setTTL(String storageGroup, long ttl) throws IOException {
	String outputStr = String.format(STRING_TYPE, MetadataOperationType.SET_TTL, storageGroup, ttl);
	ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
	channel.write(buff);
	lineNumber.incrementAndGet();
	}

	public void changeOffset(String path, long offset) throws IOException {
	String outputStr =
	String.format(STRING_TYPE, MetadataOperationType.CHANGE_OFFSET, path, offset);
	ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
	channel.write(buff);
	lineNumber.incrementAndGet();
	}

	public void changeAlias(String path, String alias) throws IOException {
	String outputStr = String.format(STRING_TYPE, MetadataOperationType.CHANGE_ALIAS, path, alias);
	ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
	channel.write(buff);
	lineNumber.incrementAndGet();
	}

	public static void upgradeMLog(String schemaDir, String logFileName) throws IOException {
	File logFile = SystemFileFactory.INSTANCE.getFile(schemaDir + File.separator + logFileName);
	File tmpLogFile = SystemFileFactory.INSTANCE.getFile(logFile.getAbsolutePath() + ".tmp");

	// if both old mlog and mlog.tmp do not exist, nothing to do
	if (!logFile.exists() && !tmpLogFile.exists()) {
	return;
	} else if (!logFile.exists() && tmpLogFile.exists()) {
	// if old mlog doesn't exsit but mlog.tmp exists, rename tmp file to mlog
	FSFactoryProducer.getFSFactory().moveFile(tmpLogFile, logFile);
	return;
	}

	// if both old mlog and mlog.tmp exist, delete mlog tmp, then do upgrading
	if (tmpLogFile.exists()) {
	try {
	Files.delete(Paths.get(tmpLogFile.toURI()));
	} catch (IOException e) {
	throw new IOException("Deleting " + tmpLogFile + "failed with exception " + e.getMessage());
	}
	}
	// upgrading
	try (BufferedReader reader = new BufferedReader(new FileReader(logFile));
	BufferedWriter writer = new BufferedWriter(new FileWriter(tmpLogFile, true))) {
	String line;
	while ((line = reader.readLine()) != null) {
	StringBuilder buf = new StringBuilder();
	buf.append(line);
	if (line.startsWith(MetadataOperationType.CREATE_TIMESERIES)) {
	buf.append(",,,");
	}
	writer.write(buf.toString());
	writer.newLine();
	writer.flush();
	}
	}
	}

	public void clear() throws IOException {
	channel.force(true);
	channel.close();
	fileOutputStream.close();
	Files.delete(logFile.toPath());
	fileOutputStream = new FileOutputStream(logFile, true);
	channel = fileOutputStream.getChannel();
	lineNumber.set(0);
	}

	public void serializeMNode(MNodePlan plan) throws IOException {
	StringBuilder s = new StringBuilder(String.valueOf(MetadataConstant.STORAGE_GROUP_MNODE_TYPE));
	s.append(",").append(plan.getName()).append(",");
	s.append(plan.getChildSize());
	s.append(LINE_SEPARATOR);
	ByteBuffer buff = ByteBuffer.wrap(s.toString().getBytes());
	channel.write(buff);
	lineNumber.incrementAndGet();
	}

	public void serializeMeasurementMNode(MeasurementMNodePlan plan) throws IOException {
	StringBuilder s = new StringBuilder(String.valueOf(MetadataConstant.MEASUREMENT_MNODE_TYPE));
	s.append(",").append(plan.getName()).append(",");
	if (plan.getAlias() != null) {
	s.append(plan.getAlias());
	}
	IMeasurementSchema schema = plan.getSchema();
	s.append(",").append(schema.getType().ordinal()).append(",");
	s.append(schema.getEncodingType().ordinal()).append(",");
	s.append(schema.getCompressor().ordinal()).append(",");
	if (schema.getProps() != null) {
	for (Map.Entry<String, String> entry : schema.getProps().entrySet()) {
	s.append(entry.getKey()).append(":").append(entry.getValue()).append(";");
	}
	}
	s.append(",").append(plan.getOffset()).append(",");
	s.append(plan.getChildSize());
	s.append(LINE_SEPARATOR);
	ByteBuffer buff = ByteBuffer.wrap(s.toString().getBytes());
	channel.write(buff);
	lineNumber.incrementAndGet();
	}

	public void serializeStorageGroupMNode(StorageGroupMNodePlan plan) throws IOException {
	StringBuilder s = new StringBuilder(String.valueOf(MetadataConstant.STORAGE_GROUP_MNODE_TYPE));
	s.append(",").append(plan.getName()).append(",");
	s.append(plan.getDataTTL()).append(",");
	s.append(plan.getChildSize());
	s.append(LINE_SEPARATOR);
	ByteBuffer buff = ByteBuffer.wrap(s.toString().getBytes());
	channel.write(buff);
	lineNumber.incrementAndGet();
	}

	public void setTemplate(SetSchemaTemplatePlan plan) throws IOException {
	StringBuilder buf = new StringBuilder(String.valueOf(MetadataOperationType.SET_TEMPLATE));
	buf.append(",");
	buf.append(plan.getTemplateName());
	buf.append(",");
	buf.append(plan.getPrefixPath());
	buf.append(LINE_SEPARATOR);
	ByteBuffer buff = ByteBuffer.wrap(buf.toString().getBytes());
	channel.write(buff);
	lineNumber.incrementAndGet();
	}

	public void setUsingTemplate(SetUsingSchemaTemplatePlan plan) throws IOException {
	StringBuilder buf = new StringBuilder(String.valueOf(MetadataOperationType.SET_USING_TEMPLATE));
	buf.append(",");
	buf.append(plan.getPrefixPath());
	buf.append(LINE_SEPARATOR);
	ByteBuffer buff = ByteBuffer.wrap(buf.toString().getBytes());
	channel.write(buff);
	lineNumber.incrementAndGet();
	}

	public void createTemplate(CreateTemplatePlan plan) throws IOException {
	StringBuilder buf = new StringBuilder();
	for (int i = 0; i < plan.getSchemaNames().size(); i++) {
	for (int j = 0; j < plan.getMeasurements().get(i).size(); j++) {
	String measurement;
	boolean isAligned = false;
	if (plan.getMeasurements().get(i).size() == 1) {
	measurement = plan.getSchemaNames().get(i);
	} else {
	// for aligned timeseries
	isAligned = true;
	measurement =
	plan.getSchemaNames().get(i)
	+ TsFileConstant.PATH_SEPARATOR
	+ plan.getMeasurements().get(i).get(j);
	}
	buf.append(
	String.format(
	"%s,%s,%s,%s,%s,%s,%s",
	MetadataOperationType.CREATE_TEMPLATE,
	plan.getName(),
	isAligned ? 1 : 0,
	measurement,
	plan.getDataTypes().get(i).get(j).serialize(),
	plan.getEncodings().get(i).get(j).serialize(),
	plan.getCompressors().get(i).serialize()));
	buf.append(LINE_SEPARATOR);
	lineNumber.incrementAndGet();
	}
	}
	ByteBuffer buff = ByteBuffer.wrap(buf.toString().getBytes());
	channel.write(buff);
	}

	public void autoCreateDeviceNode(String Device) throws IOException {
	String outputStr = MetadataOperationType.AUTO_CREATE_DEVICE + "," + Device + LINE_SEPARATOR;
	ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
	channel.write(buff);
	lineNumber.incrementAndGet();
	}

	int getLineNumber() {
	return lineNumber.get();
	}

	/** only used for initialize a mlog file writer. */
	void setLineNumber(int number) {
	lineNumber.set(number);
	}
	}