iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/AliasIndexPage.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.schemaengine.schemaregion.mtree.impl.pbtree.schemafile;

 import org.apache.iotdb.commons.exception.MetadataException;
 import org.apache.iotdb.db.exception.metadata.schemafile.RecordDuplicatedException;
 import org.apache.iotdb.db.exception.metadata.schemafile.SegmentOverflowException;
 import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;

 import java.nio.ByteBuffer;
 import java.util.ArrayDeque;
 import java.util.Queue;

 /**
  * This page stores measurement alias as key and name as record, aiming to provide a mapping from
  * alias to name so that secondary index could be built upon efficient structure.
  *
  * <p>Since it is not necessary to shift size of it, extending a {@linkplain SchemaPage} might be
  * more reasonable for this structure.
  *
  * <p>TODO: Another abstract class for this and InternalPage is expected
  */
 public class AliasIndexPage extends SchemaPage implements ISegment<String, String> {

   private static final int OFFSET_LEN = 2;
   long nextPage;
   String penultKey = null;
   String lastKey = null;

   /**
    * <b>Page Header Structure: (19 bytes used, 13 bytes reserved)</b> As in {@linkplain
    * ISchemaPage}.
    *
    * <p>Page Body Structure:
    *
    * <ul>
    *   <li>2 bytes * memberNum: offset of (alias, name) pair, to locate string content
    *   <li>... spare space...
    *   <li>var length(4 byte, var length, 4 bytes, var length): (alias, name) pairs, ordered by
    *       alias
    * </ul>
    */
   protected AliasIndexPage(ByteBuffer pageBuffer) {
     super(pageBuffer);
     nextPage = ReadWriteIOUtils.readLong(this.pageBuffer);
   }

   // region Interface Implementation

   @Override
   public int insertRecord(String alias, String name) throws RecordDuplicatedException {
     if (spareSize < OFFSET_LEN + 8 + alias.getBytes().length + name.getBytes().length) {
       return -1;
     }

     // check whether key already exists
     int pos = getIndexByKey(alias);
     if (memberNum > 0 && pos < memberNum && getKeyByIndex(pos).equals(alias)) {
       // not insert duplicated alias
       return spareSize;
     }

     if (SchemaFileConfig.PAGE_HEADER_SIZE
             + OFFSET_LEN * (memberNum + 1)
             + 8
             + alias.getBytes().length
             + name.getBytes().length
         > spareOffset) {
       compactKeys();
     }

     // append key
     this.pageBuffer.clear();
     this.spareOffset =
         (short) (this.spareOffset - alias.getBytes().length - name.getBytes().length - 8);
     this.pageBuffer.position(spareOffset);
     ReadWriteIOUtils.write(alias, this.pageBuffer);
     ReadWriteIOUtils.write(name, this.pageBuffer);

     int migNum = memberNum - pos;
     if (migNum > 0) {
       // move compound pointers
       ByteBuffer buf = ByteBuffer.allocate(migNum * OFFSET_LEN);
       this.pageBuffer.limit(SchemaFileConfig.PAGE_HEADER_SIZE + OFFSET_LEN * memberNum);
       this.pageBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + OFFSET_LEN * pos);
       buf.put(this.pageBuffer);

       this.pageBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + OFFSET_LEN * pos);
       ReadWriteIOUtils.write(spareOffset, this.pageBuffer);

       buf.clear();
       this.pageBuffer.limit(this.pageBuffer.limit() + OFFSET_LEN);
       this.pageBuffer.put(buf);
     } else {
       // append compound pointer
       this.pageBuffer.limit(this.pageBuffer.capacity());
       this.pageBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + memberNum * OFFSET_LEN);
       ReadWriteIOUtils.write(spareOffset, this.pageBuffer);
     }

     spareSize -= (alias.getBytes().length + name.getBytes().length + 8 + OFFSET_LEN);
     memberNum++;

     penultKey = lastKey;
     lastKey = alias;
     return spareSize;
   }

   @Override
   public int updateRecord(String key, String buffer) throws MetadataException {
     return -1;
   }

   @Override
   public int removeRecord(String key) {
     if (memberNum <= 0) {
       return spareSize;
     }

     int pos = getIndexByKey(key);
     if (!key.equals(getKeyByIndex(pos))) {
       return spareSize;
     }

     int migNum = memberNum - pos - 1;
     if (migNum > 0) {
       ByteBuffer temBuf = ByteBuffer.allocate(migNum * OFFSET_LEN);
       this.pageBuffer.clear().limit(SchemaFileConfig.PAGE_HEADER_SIZE + OFFSET_LEN * memberNum);
       this.pageBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + pos * OFFSET_LEN + OFFSET_LEN);
       temBuf.put(this.pageBuffer).clear();

       this.pageBuffer.clear().position(SchemaFileConfig.PAGE_HEADER_SIZE + pos * OFFSET_LEN);
       this.pageBuffer.put(temBuf);
     }
     memberNum--;
     spareSize += OFFSET_LEN + 8 + key.getBytes().length + getNameByIndex(pos).getBytes().length;

     return spareSize;
   }

   @Override
   public String getRecordByKey(String key) throws MetadataException {
     int pos = getIndexByKey(key);

     if (memberNum <= 0 || pos > memberNum) {
       return null;
     }

     // to tell whether pos is position to insert
     if (pos == 0 || pos == memberNum) {
       pos = pos == memberNum ? pos - 1 : pos;
     }

     return key.equals(getKeyByIndex(pos)) ? getNameByIndex(pos) : null;
   }

   @Override
   public String getRecordByAlias(String alias) throws MetadataException {
     return getRecordByKey(alias);
   }

   @Override
   public boolean hasRecordKey(String key) {
     int idx = getIndexByKey(key);
     return memberNum > 0 && idx < memberNum && key.equals(getKeyByIndex(idx));
   }

   @Override
   public boolean hasRecordAlias(String alias) {
     return hasRecordKey(alias);
   }

   @Override
   public Queue<String> getAllRecords() throws MetadataException {
     Queue<String> res = new ArrayDeque<>(memberNum);
     for (int i = 0; i < memberNum; i++) {
       res.add(getNameByIndex(i));
     }
     return res;
   }

   @Override
   public void syncPageBuffer() {
     super.syncPageBuffer();
     ReadWriteIOUtils.write(this.nextPage, this.pageBuffer);
   }

   @Override
   public void syncBuffer() {
     super.syncPageBuffer();
     ReadWriteIOUtils.write(this.nextPage, this.pageBuffer);
   }

   @Override
   public short size() {
     return (short) pageBuffer.capacity();
   }

   @Override
   public short getSpareSize() {
     return spareSize;
   }

   @Override
   public void delete() {
     // Do nothing
   }

   @Override
   public long getNextSegAddress() {
     return nextPage;
   }

   @Override
   public void setNextSegAddress(long nextSegAddress) {
     nextPage = nextSegAddress;
   }

   @Override
   public void extendsTo(ByteBuffer newBuffer) throws MetadataException {
     if (newBuffer.capacity() != this.pageBuffer.capacity()) {
       throw new MetadataException("AliasIndexPage can only extend to buffer with same capacity.");
     }

     syncPageBuffer();
     this.pageBuffer.clear();
     newBuffer.clear();

     newBuffer.put(this.pageBuffer);
   }

   @Override
   public String splitByKey(String key, String entry, ByteBuffer dstBuffer, boolean inclineSplit)
       throws MetadataException {

     if (this.pageBuffer.capacity() != dstBuffer.capacity()) {
       throw new MetadataException("Segments only splits with same capacity.");
     }

     if (memberNum == 0) {
       throw new MetadataException("Segment can not be split with no records.");
     }

     if (key == null && memberNum <= 1) {
       throw new MetadataException("Segment can not be split with only one record.");
     }

     // notice that key can be null here
     boolean monotonic =
         penultKey != null
             && key != null
             && lastKey != null
             && inclineSplit
             && (key.compareTo(lastKey)) * (lastKey.compareTo(penultKey)) > 0;

     int n = memberNum;

     // actual index of key just smaller than the insert, -2 for null key
     int pos = key != null ? getIndexByKey(key) : -2;

     int sp; // virtual index to split
     if (monotonic) {
       // New entry into part with more space
       sp = key.compareTo(lastKey) > 0 ? Math.max(pos, n >> 1) : Math.min(pos, n >> 1);
     } else {
       sp = n >> 1;
     }

     // Little different from InternalSegment, only the front edge key can not split
     sp = sp <= 0 ? 1 : sp;

     // This method BREAKS envelop of the passing in buffer to be more efficient
     // attributes for dstBuffer
     short memberNum = 0;
     short spareOffset = (short) dstBuffer.capacity();
     short spareSize = (short) (spareOffset - SchemaFileConfig.PAGE_HEADER_SIZE);
     dstBuffer.clear();

     short offset;
     int recSize;
     String mKey, sKey = null;
     int aix; // aix for actual index on keyAddressList
     n = key == null ? n - 1 : n; // null key
     // TODO: implement bulk split further
     for (int ix = sp; ix <= n; ix++) {
       if (ix == pos) {
         // Migrate newly insert
         mKey = key;
         recSize = 8 + entry.getBytes().length + mKey.getBytes().length;

         spareOffset -= recSize;
         dstBuffer.position(spareOffset);
         ReadWriteIOUtils.write(mKey, dstBuffer);
         ReadWriteIOUtils.write(entry, dstBuffer);

       } else {
         // Pos equals -2 if key is null
         aix = (ix > pos) && (pos != -2) ? ix - 1 : ix;
         mKey = getKeyByIndex(aix);

         offset = getOffsetByIndex(aix);
         pageBuffer.clear().position(offset);
         recSize = ReadWriteIOUtils.readInt(pageBuffer);
         pageBuffer.position(pageBuffer.position() + recSize);
         recSize += ReadWriteIOUtils.readInt(pageBuffer) + 8;
         pageBuffer.position(offset);
         pageBuffer.limit(offset + recSize);

         spareOffset -= recSize;
         dstBuffer.position(spareOffset);
         dstBuffer.put(this.pageBuffer);
       }
       dstBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + memberNum * OFFSET_LEN);
       ReadWriteIOUtils.write(spareOffset, dstBuffer);

       if (ix == sp) {
         // Search key is the first key in split segment
         sKey = mKey;
       }
       memberNum++;
       spareSize -= recSize + 2;
     }

     this.memberNum -= memberNum;
     if (key != null && sp <= pos) {
       // extra key has inserted into split buffer
       this.memberNum++;
     }

     // compact and update status
     compactKeys();
     if (key != null && sp > pos) {
       // new insert shall be in this
       if (insertRecord(key, entry) < 0) {
         throw new SegmentOverflowException(key);
       }
     }

     // flush dstBuffer header
     dstBuffer.clear();
     ReadWriteIOUtils.write(SchemaFileConfig.ALIAS_PAGE, dstBuffer);
     ReadWriteIOUtils.write(-1, dstBuffer);
     ReadWriteIOUtils.write(spareOffset, dstBuffer);
     ReadWriteIOUtils.write(spareSize, dstBuffer);
     ReadWriteIOUtils.write(memberNum, dstBuffer);
     ReadWriteIOUtils.write(nextPage, dstBuffer);

     penultKey = null;
     lastKey = null;
     return sKey;
   }

   @Override
   public ByteBuffer resetBuffer(int ptr) {
     return null;
   }

   @Override
   public String inspect() {
     syncPageBuffer();
     ByteBuffer bufferR = this.pageBuffer.asReadOnlyBuffer();
     StringBuilder builder =
         new StringBuilder(
             String.format(
                 "page_id:%d, total_seg:%d, spare_before:%d%n", pageIndex, 1, spareOffset));
     builder.append(
         String.format(
             "[AliasIndexSegment, total_alias: %d, spare_size:%d,", this.memberNum, this.spareSize));
     bufferR.clear();

     for (int i = 0; i < memberNum; i++) {
       builder.append(String.format("(%s, %s),", getKeyByIndex(i), getNameByIndex(i)));
     }
     builder.append("]\n");
     return builder.toString();
   }

   @Override
   public String toString() {
     return inspect();
   }

   @Override
   public ISegment<String, String> getAsAliasIndexPage() {
     return this;
   }

   // Endregion

   private void compactKeys() {
     ByteBuffer tempBuffer = ByteBuffer.allocate(this.pageBuffer.capacity() - this.spareOffset);
     tempBuffer.position(tempBuffer.capacity());
     this.spareOffset = (short) this.pageBuffer.capacity();
     String key;
     String name;
     int accSiz = 0;
     for (int i = 0; i < this.memberNum; i++) {
       // this.pageBuffer will not be overridden immediately
       key = getKeyByIndex(i);
       name = getNameByIndex(i);
       accSiz += key.getBytes().length + name.getBytes().length + 8;
       this.spareOffset = (short) (this.pageBuffer.capacity() - accSiz);

       this.pageBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + OFFSET_LEN * i);
       ReadWriteIOUtils.write(this.spareOffset, this.pageBuffer);

       // write tempBuffer backward
       tempBuffer.position(tempBuffer.capacity() - accSiz);
       ReadWriteIOUtils.write(key, tempBuffer);
       ReadWriteIOUtils.write(name, tempBuffer);
     }
     tempBuffer.position(tempBuffer.capacity() - accSiz);
     this.pageBuffer.position(this.spareOffset);
     this.pageBuffer.put(tempBuffer);
     this.spareSize =
         (short)
             (this.spareOffset - SchemaFileConfig.PAGE_HEADER_SIZE - OFFSET_LEN * this.memberNum);
   }

   /**
    * Binary search implementation.
    *
    * @param key to be searched or inserted.
    * @return the position of the smallest key larger than or equal to the passing in.
    */
   private int getIndexByKey(String key) {

     if (memberNum == 0 || key.compareTo(getKeyByIndex(0)) <= 0) {
       return 0;
     } else if (key.compareTo(getKeyByIndex(memberNum - 1)) >= 0) {
       return memberNum;
     }

     int head = 0;
     int tail = memberNum - 1;

     int pivot = 0;
     String pk;
     while (head != tail) {
       pivot = (head + tail) / 2;
       pk = getKeyByIndex(pivot);
       if (key.compareTo(pk) == 0) {
         return pivot;
       }

       if (key.compareTo(pk) > 0) {
         head = pivot;
       } else {
         tail = pivot;
       }

       if (head == tail - 1) {
         // notice that getKey(tail) always greater than key
         return key.compareTo(getKeyByIndex(head)) <= 0 ? head : tail;
       }
     }

     return pivot;
   }

   private short getOffsetByIndex(int index) {
     if (index < 0 || index >= memberNum) {
       // TODO: check whether reasonable to throw an unchecked
       throw new IndexOutOfBoundsException();
     }
     synchronized (pageBuffer) {
       this.pageBuffer.limit(this.pageBuffer.capacity());
       this.pageBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + index * OFFSET_LEN);
       return ReadWriteIOUtils.readShort(this.pageBuffer);
     }
   }

   private String getKeyByOffset(short offset) {
     synchronized (this.pageBuffer) {
       this.pageBuffer.limit(this.pageBuffer.capacity());
       this.pageBuffer.position(offset);
       return ReadWriteIOUtils.readString(this.pageBuffer);
     }
   }

   private String getNameByOffset(short offset) {
     synchronized (this.pageBuffer) {
       this.pageBuffer.limit(this.pageBuffer.capacity());
       this.pageBuffer.position(offset);
       this.pageBuffer.position(offset + 4 + ReadWriteIOUtils.readInt(this.pageBuffer));
       return ReadWriteIOUtils.readString(this.pageBuffer);
     }
   }

   private String getKeyByIndex(int index) {
     if (index < 0 || index > memberNum) {
       throw new IndexOutOfBoundsException();
     }
     return getKeyByOffset(getOffsetByIndex(index));
   }

   private String getNameByIndex(int index) {
     if (index < 0 || index >= memberNum) {
       throw new IndexOutOfBoundsException();
     }
     return getNameByOffset(getOffsetByIndex(index));
   }
 }
	/*
	* 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.schemaengine.schemaregion.mtree.impl.pbtree.schemafile;

	import org.apache.iotdb.commons.exception.MetadataException;
	import org.apache.iotdb.db.exception.metadata.schemafile.RecordDuplicatedException;
	import org.apache.iotdb.db.exception.metadata.schemafile.SegmentOverflowException;
	import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;

	import java.nio.ByteBuffer;
	import java.util.ArrayDeque;
	import java.util.Queue;

	/**
	* This page stores measurement alias as key and name as record, aiming to provide a mapping from
	* alias to name so that secondary index could be built upon efficient structure.
	*
	* <p>Since it is not necessary to shift size of it, extending a {@linkplain SchemaPage} might be
	* more reasonable for this structure.
	*
	* <p>TODO: Another abstract class for this and InternalPage is expected
	*/
	public class AliasIndexPage extends SchemaPage implements ISegment<String, String> {

	private static final int OFFSET_LEN = 2;
	long nextPage;
	String penultKey = null;
	String lastKey = null;

	/**
	* <b>Page Header Structure: (19 bytes used, 13 bytes reserved)</b> As in {@linkplain
	* ISchemaPage}.
	*
	* <p>Page Body Structure:
	*
	* <ul>
	* <li>2 bytes * memberNum: offset of (alias, name) pair, to locate string content
	* <li>... spare space...
	* <li>var length(4 byte, var length, 4 bytes, var length): (alias, name) pairs, ordered by
	* alias
	* </ul>
	*/
	protected AliasIndexPage(ByteBuffer pageBuffer) {
	super(pageBuffer);
	nextPage = ReadWriteIOUtils.readLong(this.pageBuffer);
	}

	// region Interface Implementation

	@Override
	public int insertRecord(String alias, String name) throws RecordDuplicatedException {
	if (spareSize < OFFSET_LEN + 8 + alias.getBytes().length + name.getBytes().length) {
	return -1;
	}

	// check whether key already exists
	int pos = getIndexByKey(alias);
	if (memberNum > 0 && pos < memberNum && getKeyByIndex(pos).equals(alias)) {
	// not insert duplicated alias
	return spareSize;
	}

	if (SchemaFileConfig.PAGE_HEADER_SIZE
	+ OFFSET_LEN * (memberNum + 1)
	+ 8
	+ alias.getBytes().length
	+ name.getBytes().length
	> spareOffset) {
	compactKeys();
	}

	// append key
	this.pageBuffer.clear();
	this.spareOffset =
	(short) (this.spareOffset - alias.getBytes().length - name.getBytes().length - 8);
	this.pageBuffer.position(spareOffset);
	ReadWriteIOUtils.write(alias, this.pageBuffer);
	ReadWriteIOUtils.write(name, this.pageBuffer);

	int migNum = memberNum - pos;
	if (migNum > 0) {
	// move compound pointers
	ByteBuffer buf = ByteBuffer.allocate(migNum * OFFSET_LEN);
	this.pageBuffer.limit(SchemaFileConfig.PAGE_HEADER_SIZE + OFFSET_LEN * memberNum);
	this.pageBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + OFFSET_LEN * pos);
	buf.put(this.pageBuffer);

	this.pageBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + OFFSET_LEN * pos);
	ReadWriteIOUtils.write(spareOffset, this.pageBuffer);

	buf.clear();
	this.pageBuffer.limit(this.pageBuffer.limit() + OFFSET_LEN);
	this.pageBuffer.put(buf);
	} else {
	// append compound pointer
	this.pageBuffer.limit(this.pageBuffer.capacity());
	this.pageBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + memberNum * OFFSET_LEN);
	ReadWriteIOUtils.write(spareOffset, this.pageBuffer);
	}

	spareSize -= (alias.getBytes().length + name.getBytes().length + 8 + OFFSET_LEN);
	memberNum++;

	penultKey = lastKey;
	lastKey = alias;
	return spareSize;
	}

	@Override
	public int updateRecord(String key, String buffer) throws MetadataException {
	return -1;
	}

	@Override
	public int removeRecord(String key) {
	if (memberNum <= 0) {
	return spareSize;
	}

	int pos = getIndexByKey(key);
	if (!key.equals(getKeyByIndex(pos))) {
	return spareSize;
	}

	int migNum = memberNum - pos - 1;
	if (migNum > 0) {
	ByteBuffer temBuf = ByteBuffer.allocate(migNum * OFFSET_LEN);
	this.pageBuffer.clear().limit(SchemaFileConfig.PAGE_HEADER_SIZE + OFFSET_LEN * memberNum);
	this.pageBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + pos * OFFSET_LEN + OFFSET_LEN);
	temBuf.put(this.pageBuffer).clear();

	this.pageBuffer.clear().position(SchemaFileConfig.PAGE_HEADER_SIZE + pos * OFFSET_LEN);
	this.pageBuffer.put(temBuf);
	}
	memberNum--;
	spareSize += OFFSET_LEN + 8 + key.getBytes().length + getNameByIndex(pos).getBytes().length;

	return spareSize;
	}

	@Override
	public String getRecordByKey(String key) throws MetadataException {
	int pos = getIndexByKey(key);

	if (memberNum <= 0 \|\| pos > memberNum) {
	return null;
	}

	// to tell whether pos is position to insert
	if (pos == 0 \|\| pos == memberNum) {
	pos = pos == memberNum ? pos - 1 : pos;
	}

	return key.equals(getKeyByIndex(pos)) ? getNameByIndex(pos) : null;
	}

	@Override
	public String getRecordByAlias(String alias) throws MetadataException {
	return getRecordByKey(alias);
	}

	@Override
	public boolean hasRecordKey(String key) {
	int idx = getIndexByKey(key);
	return memberNum > 0 && idx < memberNum && key.equals(getKeyByIndex(idx));
	}

	@Override
	public boolean hasRecordAlias(String alias) {
	return hasRecordKey(alias);
	}

	@Override
	public Queue<String> getAllRecords() throws MetadataException {
	Queue<String> res = new ArrayDeque<>(memberNum);
	for (int i = 0; i < memberNum; i++) {
	res.add(getNameByIndex(i));
	}
	return res;
	}

	@Override
	public void syncPageBuffer() {
	super.syncPageBuffer();
	ReadWriteIOUtils.write(this.nextPage, this.pageBuffer);
	}

	@Override
	public void syncBuffer() {
	super.syncPageBuffer();
	ReadWriteIOUtils.write(this.nextPage, this.pageBuffer);
	}

	@Override
	public short size() {
	return (short) pageBuffer.capacity();
	}

	@Override
	public short getSpareSize() {
	return spareSize;
	}

	@Override
	public void delete() {
	// Do nothing
	}

	@Override
	public long getNextSegAddress() {
	return nextPage;
	}

	@Override
	public void setNextSegAddress(long nextSegAddress) {
	nextPage = nextSegAddress;
	}

	@Override
	public void extendsTo(ByteBuffer newBuffer) throws MetadataException {
	if (newBuffer.capacity() != this.pageBuffer.capacity()) {
	throw new MetadataException("AliasIndexPage can only extend to buffer with same capacity.");
	}

	syncPageBuffer();
	this.pageBuffer.clear();
	newBuffer.clear();

	newBuffer.put(this.pageBuffer);
	}

	@Override
	public String splitByKey(String key, String entry, ByteBuffer dstBuffer, boolean inclineSplit)
	throws MetadataException {

	if (this.pageBuffer.capacity() != dstBuffer.capacity()) {
	throw new MetadataException("Segments only splits with same capacity.");
	}

	if (memberNum == 0) {
	throw new MetadataException("Segment can not be split with no records.");
	}

	if (key == null && memberNum <= 1) {
	throw new MetadataException("Segment can not be split with only one record.");
	}

	// notice that key can be null here
	boolean monotonic =
	penultKey != null
	&& key != null
	&& lastKey != null
	&& inclineSplit
	&& (key.compareTo(lastKey)) * (lastKey.compareTo(penultKey)) > 0;

	int n = memberNum;

	// actual index of key just smaller than the insert, -2 for null key
	int pos = key != null ? getIndexByKey(key) : -2;

	int sp; // virtual index to split
	if (monotonic) {
	// New entry into part with more space
	sp = key.compareTo(lastKey) > 0 ? Math.max(pos, n >> 1) : Math.min(pos, n >> 1);
	} else {
	sp = n >> 1;
	}

	// Little different from InternalSegment, only the front edge key can not split
	sp = sp <= 0 ? 1 : sp;

	// This method BREAKS envelop of the passing in buffer to be more efficient
	// attributes for dstBuffer
	short memberNum = 0;
	short spareOffset = (short) dstBuffer.capacity();
	short spareSize = (short) (spareOffset - SchemaFileConfig.PAGE_HEADER_SIZE);
	dstBuffer.clear();

	short offset;
	int recSize;
	String mKey, sKey = null;
	int aix; // aix for actual index on keyAddressList
	n = key == null ? n - 1 : n; // null key
	// TODO: implement bulk split further
	for (int ix = sp; ix <= n; ix++) {
	if (ix == pos) {
	// Migrate newly insert
	mKey = key;
	recSize = 8 + entry.getBytes().length + mKey.getBytes().length;

	spareOffset -= recSize;
	dstBuffer.position(spareOffset);
	ReadWriteIOUtils.write(mKey, dstBuffer);
	ReadWriteIOUtils.write(entry, dstBuffer);

	} else {
	// Pos equals -2 if key is null
	aix = (ix > pos) && (pos != -2) ? ix - 1 : ix;
	mKey = getKeyByIndex(aix);

	offset = getOffsetByIndex(aix);
	pageBuffer.clear().position(offset);
	recSize = ReadWriteIOUtils.readInt(pageBuffer);
	pageBuffer.position(pageBuffer.position() + recSize);
	recSize += ReadWriteIOUtils.readInt(pageBuffer) + 8;
	pageBuffer.position(offset);
	pageBuffer.limit(offset + recSize);

	spareOffset -= recSize;
	dstBuffer.position(spareOffset);
	dstBuffer.put(this.pageBuffer);
	}
	dstBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + memberNum * OFFSET_LEN);
	ReadWriteIOUtils.write(spareOffset, dstBuffer);

	if (ix == sp) {
	// Search key is the first key in split segment
	sKey = mKey;
	}
	memberNum++;
	spareSize -= recSize + 2;
	}

	this.memberNum -= memberNum;
	if (key != null && sp <= pos) {
	// extra key has inserted into split buffer
	this.memberNum++;
	}

	// compact and update status
	compactKeys();
	if (key != null && sp > pos) {
	// new insert shall be in this
	if (insertRecord(key, entry) < 0) {
	throw new SegmentOverflowException(key);
	}
	}

	// flush dstBuffer header
	dstBuffer.clear();
	ReadWriteIOUtils.write(SchemaFileConfig.ALIAS_PAGE, dstBuffer);
	ReadWriteIOUtils.write(-1, dstBuffer);
	ReadWriteIOUtils.write(spareOffset, dstBuffer);
	ReadWriteIOUtils.write(spareSize, dstBuffer);
	ReadWriteIOUtils.write(memberNum, dstBuffer);
	ReadWriteIOUtils.write(nextPage, dstBuffer);

	penultKey = null;
	lastKey = null;
	return sKey;
	}

	@Override
	public ByteBuffer resetBuffer(int ptr) {
	return null;
	}

	@Override
	public String inspect() {
	syncPageBuffer();
	ByteBuffer bufferR = this.pageBuffer.asReadOnlyBuffer();
	StringBuilder builder =
	new StringBuilder(
	String.format(
	"page_id:%d, total_seg:%d, spare_before:%d%n", pageIndex, 1, spareOffset));
	builder.append(
	String.format(
	"[AliasIndexSegment, total_alias: %d, spare_size:%d,", this.memberNum, this.spareSize));
	bufferR.clear();

	for (int i = 0; i < memberNum; i++) {
	builder.append(String.format("(%s, %s),", getKeyByIndex(i), getNameByIndex(i)));
	}
	builder.append("]\n");
	return builder.toString();
	}

	@Override
	public String toString() {
	return inspect();
	}

	@Override
	public ISegment<String, String> getAsAliasIndexPage() {
	return this;
	}

	// Endregion

	private void compactKeys() {
	ByteBuffer tempBuffer = ByteBuffer.allocate(this.pageBuffer.capacity() - this.spareOffset);
	tempBuffer.position(tempBuffer.capacity());
	this.spareOffset = (short) this.pageBuffer.capacity();
	String key;
	String name;
	int accSiz = 0;
	for (int i = 0; i < this.memberNum; i++) {
	// this.pageBuffer will not be overridden immediately
	key = getKeyByIndex(i);
	name = getNameByIndex(i);
	accSiz += key.getBytes().length + name.getBytes().length + 8;
	this.spareOffset = (short) (this.pageBuffer.capacity() - accSiz);

	this.pageBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + OFFSET_LEN * i);
	ReadWriteIOUtils.write(this.spareOffset, this.pageBuffer);

	// write tempBuffer backward
	tempBuffer.position(tempBuffer.capacity() - accSiz);
	ReadWriteIOUtils.write(key, tempBuffer);
	ReadWriteIOUtils.write(name, tempBuffer);
	}
	tempBuffer.position(tempBuffer.capacity() - accSiz);
	this.pageBuffer.position(this.spareOffset);
	this.pageBuffer.put(tempBuffer);
	this.spareSize =
	(short)
	(this.spareOffset - SchemaFileConfig.PAGE_HEADER_SIZE - OFFSET_LEN * this.memberNum);
	}

	/**
	* Binary search implementation.
	*
	* @param key to be searched or inserted.
	* @return the position of the smallest key larger than or equal to the passing in.
	*/
	private int getIndexByKey(String key) {

	if (memberNum == 0 \|\| key.compareTo(getKeyByIndex(0)) <= 0) {
	return 0;
	} else if (key.compareTo(getKeyByIndex(memberNum - 1)) >= 0) {
	return memberNum;
	}

	int head = 0;
	int tail = memberNum - 1;

	int pivot = 0;
	String pk;
	while (head != tail) {
	pivot = (head + tail) / 2;
	pk = getKeyByIndex(pivot);
	if (key.compareTo(pk) == 0) {
	return pivot;
	}

	if (key.compareTo(pk) > 0) {
	head = pivot;
	} else {
	tail = pivot;
	}

	if (head == tail - 1) {
	// notice that getKey(tail) always greater than key
	return key.compareTo(getKeyByIndex(head)) <= 0 ? head : tail;
	}
	}

	return pivot;
	}

	private short getOffsetByIndex(int index) {
	if (index < 0 \|\| index >= memberNum) {
	// TODO: check whether reasonable to throw an unchecked
	throw new IndexOutOfBoundsException();
	}
	synchronized (pageBuffer) {
	this.pageBuffer.limit(this.pageBuffer.capacity());
	this.pageBuffer.position(SchemaFileConfig.PAGE_HEADER_SIZE + index * OFFSET_LEN);
	return ReadWriteIOUtils.readShort(this.pageBuffer);
	}
	}

	private String getKeyByOffset(short offset) {
	synchronized (this.pageBuffer) {
	this.pageBuffer.limit(this.pageBuffer.capacity());
	this.pageBuffer.position(offset);
	return ReadWriteIOUtils.readString(this.pageBuffer);
	}
	}

	private String getNameByOffset(short offset) {
	synchronized (this.pageBuffer) {
	this.pageBuffer.limit(this.pageBuffer.capacity());
	this.pageBuffer.position(offset);
	this.pageBuffer.position(offset + 4 + ReadWriteIOUtils.readInt(this.pageBuffer));
	return ReadWriteIOUtils.readString(this.pageBuffer);
	}
	}

	private String getKeyByIndex(int index) {
	if (index < 0 \|\| index > memberNum) {
	throw new IndexOutOfBoundsException();
	}
	return getKeyByOffset(getOffsetByIndex(index));
	}

	private String getNameByIndex(int index) {
	if (index < 0 \|\| index >= memberNum) {
	throw new IndexOutOfBoundsException();
	}
	return getNameByOffset(getOffsetByIndex(index));
	}
	}