table.java

来自「RESIN 3.2 最新源码」· Java 代码 · 共 912 行 · 第 1/2 页

	while (iter.nextRow()) {
	  long rowAddress = iter.getRowAddress();
	  int rowOffset = iter.getRowOffset();
	  
	  for (int i = 0; i < columns.length; i++) {
	    Column column = columns[i];

	    column.setIndex(xa, blockBuffer, rowOffset, rowAddress, null);
	  }
	}
      }
    } finally {
      xa.commit();
    }
  }

  private void writeTableHeader(WriteStream os)
    throws IOException
  {
    os.print(DB_VERSION);
    os.write(0);

    while (os.getPosition() < INDEX_ROOT_OFFSET) {
      os.write(0);
    }

    Column []columns = _row.getColumns();
    for (int i = 0; i < columns.length; i++) {
      if (! columns[i].isUnique())
	continue;

      BTree index = columns[i].getIndex();

      if (index != null) {
	writeLong(os, index.getIndexRoot());
      }
      else {
	writeLong(os, 0);
      }
    }

    while (os.getPosition() < ROOT_DATA_END) {
      os.write(0);
    }

    os.print("CREATE TABLE " + getName() + "(");
    for (int i = 0; i < _row.getColumns().length; i++) {
      Column column = _row.getColumns()[i];
      
      if (i != 0)
	os.print(",");
      
      os.print(column.getName());
      os.print(" ");

      switch (column.getTypeCode()) {
      case Column.VARCHAR:
	os.print("VARCHAR(" + column.getDeclarationSize() + ")");
	break;
      case Column.VARBINARY:
	os.print("VARBINARY(" + column.getDeclarationSize() + ")");
	break;
      case Column.BINARY:
	os.print("BINARY(" + column.getDeclarationSize() + ")");
	break;
      case Column.SHORT:
	os.print("SMALLINT");
	break;
      case Column.INT:
	os.print("INTEGER");
	break;
      case Column.LONG:
	os.print("BIGINT");
	break;
      case Column.DOUBLE:
	os.print("DOUBLE");
	break;
      case Column.DATE:
	os.print("TIMESTAMP");
	break;
      case Column.BLOB:
	os.print("BLOB");
	break;
      case Column.NUMERIC:
	{
	  NumericColumn numeric = (NumericColumn) column;
	  
	  os.print("NUMERIC(" + numeric.getPrecision() + "," + numeric.getScale() + ")");
	  break;
	}
      default:
	throw new UnsupportedOperationException();
      }

      if (column.isPrimaryKey())
	os.print(" PRIMARY KEY");
      else if (column.isUnique())
	os.print(" UNIQUE");
      
      if (column.isNotNull())
	os.print(" NOT NULL");

      Expr defaultExpr = column.getDefault();
      
      if (defaultExpr != null) {
	os.print(" DEFAULT (");
	os.print(defaultExpr);
	os.print(")");
      }

      if (column.getAutoIncrement() >= 0)
	os.print(" auto_increment");
    }
    os.print(")");

    /*
    writeLong(os, _blockMax);
    writeLong(os, _entries);
    writeLong(os, _clockAddr);
    */
  }

  public TableIterator createTableIterator()
  {
    assertStoreActive();
    
    return new TableIterator(this);
  }

  /**
   * Returns the next auto-increment value.
   */
  public long nextAutoIncrement(QueryContext context)
    throws SQLException
  {
    synchronized (this) {
      if (_autoIncrementValue >= 0)
	return ++_autoIncrementValue;
    }

    long max = 0;
    
    try {
      TableIterator iter = createTableIterator();
      iter.init(context);
      while (iter.next()) {
	byte []buffer = iter.getBuffer();
      
	long value = _autoIncrementColumn.getLong(buffer, iter.getRowOffset());

	if (max < value)
	  max = value;
      }
    } catch (IOException e) {
      throw new SQLExceptionWrapper(e);
    }

    synchronized (this) {
      if (_autoIncrementValue < max)
	_autoIncrementValue = max;

      return ++_autoIncrementValue;
    }
  }

  /**
   * Inserts a new row, returning the row address.
   */
  public long insert(QueryContext queryContext, Transaction xa,
		     ArrayList<Column> columns,
		     ArrayList<Expr> values)
    throws IOException, SQLException
  {
    if (log.isLoggable(Level.FINEST))
      log.finest("db table " + getName() + " insert row xa:" + xa);
    
    Block block = null;
    
    try {
      long addr;
      int rowOffset = 0;
      
      boolean isLoop = false;
      boolean hasRow = false;

      int rowClockCount = 0;
      long rowClockAddr = 0;
      long rowClockUsed = 0;
      long rowClockTotal = 0;
      
      do {
	long blockId = 0;

	if (block != null) {
	  block.free();
	  block = null;
	}

	synchronized (_rowClockLock) {
	  blockId = firstRow(_rowClockAddr);

	  if (blockId >= 0) {
	  }
	  else if (! isLoop
		   && (ROW_CLOCK_MIN < _rowClockTotal
		       && 4 * _rowClockUsed < 3 * _rowClockTotal
		       || _rowAllocCount > 8)) {
	    // System.out.println("LOOP: used:" + _rowClockUsed + " total:" + _rowClockTotal + " frac:" + (double) _rowClockUsed / (double) (_rowClockTotal + 0.01));
	    // go around loop if there are sufficient entries, i.e. over
	    // ROW_CLOCK_MIN and at least 1/4 free entries.
	    isLoop = true;
	    _rowClockCount = 0;
	    _rowClockAddr = 0;
	    _rowClockUsed = 0;
	    _rowClockTotal = 0;
	    _rowAllocCount = 0;
	    continue;
	  }
	  else {
	    //System.out.println("ROW: used:" + _rowClockUsed + " total:" + _rowClockTotal + " frac:" + (double) _rowClockUsed / (double) (_rowClockTotal + 0.01));

	    _rowAllocCount++;
	    
	    // if no free row is available, allocate a new one
	    block = xa.allocateRow(this);
	    //System.out.println("ALLOC: " + block);

	    blockId = block.getBlockId();
	  }

	  rowClockCount = _rowClockCount;
	  rowClockAddr = blockIdToAddress(blockId);
	  rowClockUsed = _rowClockUsed;
	  rowClockTotal = _rowClockTotal;

	  // the next insert will try the following block
	  _rowClockCount++;
	  _rowClockAddr = rowClockAddr + BLOCK_SIZE;
	  _rowClockUsed = rowClockUsed + _rowsPerBlock;
	  _rowClockTotal = rowClockTotal + _rowsPerBlock;
	}

	if (block == null)
	  block = xa.readBlock(this, blockId);

	Lock blockLock = block.getLock();

	if (xa.lockReadAndWriteNoWait(blockLock)) {
	  try {
	    rowOffset = 0;

	    byte []buffer = block.getBuffer();
	    
	    for (; rowOffset < _rowEnd; rowOffset += _rowLength) {
	      if (buffer[rowOffset] == 0) {
		block.setDirty(rowOffset, rowOffset + 1);
			     
		hasRow = true;
		buffer[rowOffset] = ROW_ALLOC;
		break;
	      }
	    }
	  } finally {
	    xa.unlockReadAndWrite(blockLock);
	  }
	}
      } while (! hasRow);

      insertRow(queryContext, xa, columns, values,
		block, rowOffset);

      synchronized (_rowClockLock) {
	if (rowClockCount < _rowClockCount) {
	  // the next insert will retry this block
	  int blocks = _rowClockCount - rowClockCount;

	  _rowClockCount = rowClockCount;
	  _rowClockAddr = rowClockAddr;
	  _rowClockUsed -= blocks * _rowsPerBlock;
	  _rowClockTotal -= blocks * _rowsPerBlock;
	}
      }

      return blockIdToAddress(block.getBlockId(), rowOffset);
    } finally {
      if (block != null)
	block.free();
    }
  }

  public void insertRow(QueryContext queryContext, Transaction xa,
			ArrayList<Column> columns,
			ArrayList<Expr> values,
			Block block, int rowOffset)
    throws SQLException
  {
    byte []buffer = block.getBuffer();

    long rowAddr = blockIdToAddress(block.getBlockId(), rowOffset);
    //System.out.println("ADDR:" + rowAddr + " " + rowOffset + " " + block);

    TableIterator iter = createTableIterator();
    TableIterator []iterSet = new TableIterator[] { iter };
    // QueryContext context = QueryContext.allocate();
    queryContext.init(xa, iterSet, true);
    iter.init(queryContext);

    boolean isOkay = false;
    queryContext.lock();
    try {
      iter.setRow(block, rowOffset);

      block.setDirty(rowOffset, rowOffset + _rowLength);
	  
      for (int i = rowOffset + _rowLength - 1; rowOffset < i; i--)
	buffer[i] = 0;

      for (int i = 0; i < columns.size(); i++) {
	Column column = columns.get(i);
	Expr value = values.get(i);

	column.setExpr(xa, buffer, rowOffset, value, queryContext);
      }

      // lock for insert, i.e. entries, indices, and validation
      // XXX: the set index needs to handle the validation
      //xa.lockWrite(_insertLock);
      try {
	validate(block, rowOffset, queryContext, xa);
      
	buffer[rowOffset] = (byte) ((buffer[rowOffset] & ~ROW_MASK) | ROW_VALID);

	for (int i = 0; i < columns.size(); i++) {
	  Column column = columns.get(i);
	  Expr value = values.get(i);

	  column.setIndex(xa, buffer, rowOffset, rowAddr, queryContext);
	}

	xa.addUpdateBlock(block);

	if (_autoIncrementColumn != null) {
	  long value = _autoIncrementColumn.getLong(buffer, rowOffset);

	  synchronized (this) {
	    if (_autoIncrementValue < value)
	      _autoIncrementValue = value;
	  }
	}
	  
	_entries++;
      
	isOkay = true;
      } finally {
	// xa.unlockWrite(_insertLock);
	  
	if (! isOkay)
	  delete(xa, block, buffer, rowOffset);
      }
    } finally {
      queryContext.unlock();
    }
  }

  /**
   * Validates the given row.
   */
  private void validate(Block block, int rowOffset,
			QueryContext queryContext, Transaction xa)
    throws SQLException
  {
    TableIterator row = createTableIterator();
    TableIterator []rows = new TableIterator[] { row };

    row.setRow(block, rowOffset);

    for (int i = 0; i < _constraints.length; i++) {
      _constraints[i].validate(rows, queryContext, xa);
    }
  }
  
  void delete(Transaction xa, Block block, byte []buffer, int rowOffset)
    throws SQLException
  {
    byte rowState = buffer[rowOffset];

    if ((rowState & ROW_MASK) != ROW_VALID)
      return;
    
    buffer[rowOffset] = (byte) ((rowState & ~ROW_MASK) | ROW_ALLOC);
    
    Column []columns = _row.getColumns();
    
    for (int i = 0; i < columns.length; i++) {
      columns[i].delete(xa, buffer, rowOffset);
    }

    buffer[rowOffset] = 0;
    
    synchronized (_rowClockLock) {
      long addr = blockIdToAddress(block.getBlockId());
      
      if (addr <= _rowClockAddr) {
	_rowClockUsed--;
      }
    }
  }

  private void writeLong(WriteStream os, long value)
    throws IOException
  {
    os.write((int) (value >> 56));
    os.write((int) (value >> 48));
    os.write((int) (value >> 40));
    os.write((int) (value >> 32));
    os.write((int) (value >> 24));
    os.write((int) (value >> 16));
    os.write((int) (value >> 8));
    os.write((int) value);
  }

  private void setLong(byte []buffer, int offset, long value)
    throws IOException
  {
    buffer[offset + 0] = (byte) (value >> 56);
    buffer[offset + 1] = (byte) (value >> 48);
    buffer[offset + 2] = (byte) (value >> 40);
    buffer[offset + 3] = (byte) (value >> 32);
    buffer[offset + 4] = (byte) (value >> 24);
    buffer[offset + 5] = (byte) (value >> 16);
    buffer[offset + 6] = (byte) (value >> 8);
    buffer[offset + 7] = (byte) (value);
  }

  private long getLong(byte []buffer, int offset)
    throws IOException
  {
    long value = (((buffer[offset + 0] & 0xffL) << 56) +
		  ((buffer[offset + 1] & 0xffL) << 48) +
		  ((buffer[offset + 2] & 0xffL) << 40) +
		  ((buffer[offset + 3] & 0xffL) << 32) +

		  ((buffer[offset + 4] & 0xffL) << 24) +
		  ((buffer[offset + 5] & 0xffL) << 16) +
		  ((buffer[offset + 6] & 0xffL) << 8) +
		  ((buffer[offset + 7] & 0xffL)));

    return value;
  }
  
  public String toString()
  {
    return "Table[" + getName() + ":" + getId() + "]";
  }
}