ndbdictionaryimpl.cpp

mysql-5.0.22.tar.gz源码包

    n--;
    char btname[NdbBlobImpl::BlobTableNameSize];
    NdbBlob::getBlobTableName(btname, &t, &c);
    // Save BLOB table handle
    NdbTableImpl * cachedBlobTable = getTable(btname);
    if (cachedBlobTable == 0) {
      return -1;
    }
    c.m_blobTable = cachedBlobTable;
  }
  
  return 0;
}

int 
NdbDictInterface::createTable(Ndb & ndb,
			      NdbTableImpl & impl)
{
  return createOrAlterTable(ndb, impl, false);
}

int NdbDictionaryImpl::alterTable(NdbTableImpl &impl)
{
  BaseString internalName(impl.m_internalName);
  const char * originalInternalName = internalName.c_str();

  DBUG_ENTER("NdbDictionaryImpl::alterTable");
  Ndb_local_table_info * local = 0;
  if((local= get_local_table_info(originalInternalName, false)) == 0)
  {
    m_error.code = 709;
    DBUG_RETURN(-1);
  }

  // Alter the table
  int ret = m_receiver.alterTable(m_ndb, impl);
  if(ret == 0){
    // Remove cached information and let it be refreshed at next access
    m_globalHash->lock();
    local->m_table_impl->m_status = NdbDictionary::Object::Invalid;
    m_globalHash->drop(local->m_table_impl);
    m_globalHash->unlock();
    m_localHash.drop(originalInternalName);
  }
  DBUG_RETURN(ret);
}

int 
NdbDictInterface::alterTable(Ndb & ndb,
			      NdbTableImpl & impl)
{
  return createOrAlterTable(ndb, impl, true);
}

int 
NdbDictInterface::createOrAlterTable(Ndb & ndb,
				     NdbTableImpl & impl,
				     bool alter)
{
  DBUG_ENTER("NdbDictInterface::createOrAlterTable");
  unsigned i, err;
  if((unsigned)impl.getNoOfPrimaryKeys() > NDB_MAX_NO_OF_ATTRIBUTES_IN_KEY){
    m_error.code= 4317;
    DBUG_RETURN(-1);
  }
  unsigned sz = impl.m_columns.size();
  if (sz > NDB_MAX_ATTRIBUTES_IN_TABLE){
    m_error.code= 4318;
    DBUG_RETURN(-1);
  }

  if (!impl.m_newExternalName.empty()) {
    impl.m_externalName.assign(impl.m_newExternalName);
    AlterTableReq::setNameFlag(impl.m_changeMask, true);
  }

  //validate();
  //aggregate();

  const BaseString internalName(
    ndb.internalize_table_name(impl.m_externalName.c_str()));
  impl.m_internalName.assign(internalName);
  UtilBufferWriter w(m_buffer);
  DictTabInfo::Table tmpTab; tmpTab.init();
  BaseString::snprintf(tmpTab.TableName,
	   sizeof(tmpTab.TableName),
	   internalName.c_str());

  bool haveAutoIncrement = false;
  Uint64 autoIncrementValue = 0;
  Uint32 distKeys= 0;
  for(i = 0; i<sz; i++){
    const NdbColumnImpl * col = impl.m_columns[i];
    if(col == 0)
      continue;
    if (col->m_autoIncrement) {
      if (haveAutoIncrement) {
        m_error.code= 4335;
        DBUG_RETURN(-1);
      }
      haveAutoIncrement = true;
      autoIncrementValue = col->m_autoIncrementInitialValue;
    }
    if (col->m_distributionKey)
      distKeys++;
  }

  // Check max length of frm data
  if (impl.m_frm.length() > MAX_FRM_DATA_SIZE){
    m_error.code= 1229;
    DBUG_RETURN(-1);
  }
  tmpTab.FrmLen = impl.m_frm.length();
  memcpy(tmpTab.FrmData, impl.m_frm.get_data(), impl.m_frm.length());

  tmpTab.TableLoggedFlag = impl.m_logging;
  tmpTab.TableKValue = impl.m_kvalue;
  tmpTab.MinLoadFactor = impl.m_minLoadFactor;
  tmpTab.MaxLoadFactor = impl.m_maxLoadFactor;
  tmpTab.TableType = DictTabInfo::UserTable;
  tmpTab.NoOfAttributes = sz;
  
  tmpTab.FragmentType = getKernelConstant(impl.m_fragmentType,
					  fragmentTypeMapping,
					  DictTabInfo::AllNodesSmallTable);
  tmpTab.TableVersion = rand();

  SimpleProperties::UnpackStatus s;
  s = SimpleProperties::pack(w, 
			     &tmpTab,
			     DictTabInfo::TableMapping, 
			     DictTabInfo::TableMappingSize, true);
  
  if(s != SimpleProperties::Eof){
    abort();
  }
  
  if (distKeys == impl.m_noOfKeys)
    distKeys= 0;
  impl.m_noOfDistributionKeys= distKeys;
  
  for(i = 0; i<sz; i++){
    const NdbColumnImpl * col = impl.m_columns[i];
    if(col == 0)
      continue;
    
    DictTabInfo::Attribute tmpAttr; tmpAttr.init();
    BaseString::snprintf(tmpAttr.AttributeName, sizeof(tmpAttr.AttributeName), 
	     col->m_name.c_str());
    tmpAttr.AttributeId = i;
    tmpAttr.AttributeKeyFlag = col->m_pk;
    tmpAttr.AttributeNullableFlag = col->m_nullable;
    tmpAttr.AttributeDKey = distKeys ? col->m_distributionKey : 0;

    tmpAttr.AttributeExtType = (Uint32)col->m_type;
    tmpAttr.AttributeExtPrecision = ((unsigned)col->m_precision & 0xFFFF);
    tmpAttr.AttributeExtScale = col->m_scale;
    tmpAttr.AttributeExtLength = col->m_length;

    // check type and compute attribute size and array size
    if (! tmpAttr.translateExtType()) {
      m_error.code= 703;
      DBUG_RETURN(-1);
    }
    // charset is defined exactly for char types
    if (col->getCharType() != (col->m_cs != NULL)) {
      m_error.code= 703;
      DBUG_RETURN(-1);
    }
    // primary key type check
    if (col->m_pk && 
        (err = NdbSqlUtil::check_column_for_pk(col->m_type, col->m_cs)))
    {
      m_error.code= err;
      DBUG_RETURN(-1);
    }
    // distribution key not supported for Char attribute
    if (distKeys && col->m_distributionKey && col->m_cs != NULL) {
      m_error.code= 745;
      DBUG_RETURN(-1);
    }
    // charset in upper half of precision
    if (col->getCharType()) {
      tmpAttr.AttributeExtPrecision |= (col->m_cs->number << 16);
    }

    tmpAttr.AttributeAutoIncrement = col->m_autoIncrement;
    BaseString::snprintf(tmpAttr.AttributeDefaultValue, 
	     sizeof(tmpAttr.AttributeDefaultValue),
	     col->m_defaultValue.c_str());
    s = SimpleProperties::pack(w, 
			       &tmpAttr,
			       DictTabInfo::AttributeMapping, 
			       DictTabInfo::AttributeMappingSize, true);
    w.add(DictTabInfo::AttributeEnd, 1);
  }

  NdbApiSignal tSignal(m_reference);
  tSignal.theReceiversBlockNumber = DBDICT;
  
  LinearSectionPtr ptr[1];
  ptr[0].p = (Uint32*)m_buffer.get_data();
  ptr[0].sz = m_buffer.length() / 4;
  int ret;
  if (alter)
  {
    AlterTableReq * const req = 
      CAST_PTR(AlterTableReq, tSignal.getDataPtrSend());
    
    req->senderRef = m_reference;
    req->senderData = 0;
    req->changeMask = impl.m_changeMask;
    req->tableId = impl.m_tableId;
    req->tableVersion = impl.m_version;;
    tSignal.theVerId_signalNumber   = GSN_ALTER_TABLE_REQ;
    tSignal.theLength = AlterTableReq::SignalLength;
    ret= alterTable(&tSignal, ptr);
  }
  else
  {
    CreateTableReq * const req = 
      CAST_PTR(CreateTableReq, tSignal.getDataPtrSend());
    
    req->senderRef = m_reference;
    req->senderData = 0;
    tSignal.theVerId_signalNumber   = GSN_CREATE_TABLE_REQ;
    tSignal.theLength = CreateTableReq::SignalLength;
    ret= createTable(&tSignal, ptr);

    if (ret)
      DBUG_RETURN(ret);

    if (haveAutoIncrement) {
      if (!ndb.setAutoIncrementValue(impl.m_externalName.c_str(),
				     autoIncrementValue)) {
	if (ndb.theError.code == 0) {
	  m_error.code= 4336;
	  ndb.theError = m_error;
	} else
	  m_error= ndb.theError;
	ret = -1; // errorcode set in initialize_autoincrement
      }
    }
  }
  DBUG_RETURN(ret);
}

int
NdbDictInterface::createTable(NdbApiSignal* signal, LinearSectionPtr ptr[3])
{
#if DEBUG_PRINT
  ndbout_c("BufferLen = %d", ptr[0].sz);
  SimplePropertiesLinearReader r(ptr[0].p, ptr[0].sz);
  r.printAll(ndbout);
#endif
  const int noErrCodes = 2;
  int errCodes[noErrCodes] = 
     {CreateTableRef::Busy,
      CreateTableRef::NotMaster};
  return dictSignal(signal,ptr,1,
		    1/*use masternode id*/,
		    100,
		    WAIT_CREATE_INDX_REQ,
		    WAITFOR_RESPONSE_TIMEOUT,
		    errCodes,noErrCodes);
}


void
NdbDictInterface::execCREATE_TABLE_CONF(NdbApiSignal * signal,
					LinearSectionPtr ptr[3])
{
#if 0
  const CreateTableConf* const conf=
    CAST_CONSTPTR(CreateTableConf, signal->getDataPtr());
  Uint32 tableId= conf->tableId;
  Uint32 tableVersion= conf->tableVersion;
#endif
  m_waiter.signal(NO_WAIT);  
}

void
NdbDictInterface::execCREATE_TABLE_REF(NdbApiSignal * signal,
				       LinearSectionPtr ptr[3])
{
  const CreateTableRef* const ref=
    CAST_CONSTPTR(CreateTableRef, signal->getDataPtr());
  m_error.code= ref->errorCode;
  m_masterNodeId = ref->masterNodeId;
  m_waiter.signal(NO_WAIT);  
}

int
NdbDictInterface::alterTable(NdbApiSignal* signal, LinearSectionPtr ptr[3])
{
#if DEBUG_PRINT
  ndbout_c("BufferLen = %d", ptr[0].sz);
  SimplePropertiesLinearReader r(ptr[0].p, ptr[0].sz);
  r.printAll(ndbout);
#endif
  const int noErrCodes = 2;
  int errCodes[noErrCodes] =
    {AlterTableRef::NotMaster,
     AlterTableRef::Busy};
  int r = dictSignal(signal,ptr,1,
		     1/*use masternode id*/,
		     100,WAIT_ALTER_TAB_REQ,
		     WAITFOR_RESPONSE_TIMEOUT,
		     errCodes, noErrCodes);
  if(m_error.code == AlterTableRef::InvalidTableVersion) {
    // Clear caches and try again
    return INCOMPATIBLE_VERSION;
  }

  return r;
}

void
NdbDictInterface::execALTER_TABLE_CONF(NdbApiSignal * signal,
                                       LinearSectionPtr ptr[3])
{
  //AlterTableConf* const conf = CAST_CONSTPTR(AlterTableConf, signal->getDataPtr());
  m_waiter.signal(NO_WAIT);
}

void
NdbDictInterface::execALTER_TABLE_REF(NdbApiSignal * signal,
				      LinearSectionPtr ptr[3])
{
  const AlterTableRef * const ref = 
    CAST_CONSTPTR(AlterTableRef, signal->getDataPtr());
  m_error.code= ref->errorCode;
  m_masterNodeId = ref->masterNodeId;
  m_waiter.signal(NO_WAIT);
}

/*****************************************************************
 * Drop table
 */
int
NdbDictionaryImpl::dropTable(const char * name)
{
  DBUG_ENTER("NdbDictionaryImpl::dropTable");
  DBUG_PRINT("enter",("name: %s", name));
  NdbTableImpl * tab = getTable(name);
  if(tab == 0){
    DBUG_RETURN(-1);
  }
  int ret = dropTable(* tab);
  // If table stored in cache is incompatible with the one in the kernel
  // we must clear the cache and try again
  if (ret == INCOMPATIBLE_VERSION) {
    const BaseString internalTableName(m_ndb.internalize_table_name(name));

    DBUG_PRINT("info",("INCOMPATIBLE_VERSION internal_name: %s", internalTableName.c_str()));
    m_localHash.drop(internalTableName.c_str());
    m_globalHash->lock();
    tab->m_status = NdbDictionary::Object::Invalid;
    m_globalHash->drop(tab);
    m_globalHash->unlock();
    DBUG_RETURN(dropTable(name));
  }

  DBUG_RETURN(ret);
}

int
NdbDictionaryImpl::dropTable(NdbTableImpl & impl)
{
  int res;
  const char * name = impl.getName();
  if(impl.m_status == NdbDictionary::Object::New){
    return dropTable(name);
  }

  if (impl.m_indexType != NdbDictionary::Index::Undefined) {
    m_receiver.m_error.code= 1228;
    return -1;
  }

  List list;
  if ((res = listIndexes(list, impl.m_tableId)) == -1){
    return -1;
  }
  for (unsigned i = 0; i < list.count; i++) {
    const List::Element& element = list.elements[i];
    if ((res = dropIndex(element.name, name)) == -1)
    {
      return -1;
    }
  }
  
  if (impl.m_noOfBlobs != 0) {
    if (dropBlobTables(impl) != 0){
      return -1;
    }
  }
  
  int ret = m_receiver.dropTable(impl);  
  if(ret == 0 || m_error.code == 709){
    const char * internalTableName = impl.m_internalName.c_str();

    
    m_localHash.drop(internalTableName);
    m_globalHash->lock();
    impl.m_status = NdbDictionary::Object::Invalid;
    m_globalHash->drop(&impl);
    m_globalHash->unlock();

    return 0;
  }
  
  return ret;
}

int
NdbDictionaryImpl::dropBlobTables(NdbTableImpl & t)
{
  DBUG_ENTER("NdbDictionaryImpl::dropBlobTables");
  for (unsigned i = 0; i < t.m_columns.size(); i++) {
    NdbColumnImpl & c = *t.m_columns[i];
    if (! c.getBlobType() || c.getPartSize() == 0)
      continue;
    char btname[NdbBlobImpl::BlobTableNameSize];
    NdbBlob::getBlobTableName(btname, &t, &c);
    if (dropTable(btname) != 0) {
      if (m_error.code != 709){
	DBUG_PRINT("exit",("error %u - exiting",m_error.code));
        DBUG_RETURN(-1);
      }
      DBUG_PRINT("info",("error %u - continuing",m_error.code));
    }
  }
  DBUG_RETURN(0);
}

int
NdbDictInterface::dropTable(const NdbTableImpl & impl)
{
  NdbApiSignal tSignal(m_reference);
  tSignal.theReceiversBlockNumber = DBDICT;
  tSignal.theVerId_signalNumber   = GSN_DROP_TABLE_REQ;
  tSignal.theLength = DropTableReq::SignalLength;
  
  DropTableReq * const req = CAST_PTR(DropTableReq, tSignal.getDataPtrSend());
  req->senderRef = m_reference;
  req->senderData = 0;
  req->tableId = impl.m_tableId;
  req->tableVersion = impl.m_version;

  return dropTable(&tSignal, 0);
}

int
NdbDictInterface::dropTable(NdbApiSignal* signal, LinearSectionPtr ptr[3])
{
  const int noErrCodes = 3;
  int errCodes[noErrCodes] =
         {DropTableRef::NoDropTableRecordAvailable,
          DropTableRef::NotMaster,
          DropTableRef::Busy};
  int r = dictSignal(signal,NULL,0,
		     1/*use masternode id*/,
		     100,WAIT_DROP_TAB_REQ,
		     WAITFOR_RESPONSE_TIMEOUT,
		     errCodes, noErrCodes);
  if(m_error.code == DropTableRef::InvalidTableVersion) {
    // Clear caches and try again
    return INCOMPATIBLE_VERSION;
  }
  return r;
}

void
NdbDictInterface::execDROP_TABLE_CONF(NdbApiSignal * signal,
				       LinearSectionPtr ptr[3])
{
  DBUG_ENTER("NdbDictInterface::execDROP_TABLE_CONF");
  //DropTableConf* const conf = CAST_CONSTPTR(DropTableConf, signal->getDataPtr());

  m_waiter.signal(NO_WAIT);  
  DBUG_VOID_RETURN;
}

void
NdbDictInterface::execDROP_TABLE_REF(NdbApiSignal * signal,
				      LinearSectionPtr ptr[3])
{
  DBUG_ENTER("NdbDictInterface::execDROP_TABLE_REF");
  const DropTableRef* const ref = CAST_CONSTPTR(DropTableRef, signal->getDataPtr());
  m_error.code= ref->errorCode;
  m_masterNodeId = ref->masterNodeId;
  m_waiter.signal(NO_WAIT);  
  DBUG_VOID_RETURN;
}

int
NdbDictionaryImpl::invalidateObject(NdbTableImpl & impl)
{
  const char * internalTableName = impl.m_internalName.c_str();
  DBUG_ENTER("NdbDictionaryImpl::invalidateObject");
  DBUG_PRINT("enter", ("internal_name: %s", internalTableName));
  m_localHash.drop(internalTableName);
  m_globalHash->lock();
  impl.m_status = NdbDictionary::Object::Invalid;