4493d40a-92aa-4162-9dfc-bfc4ec0996e6
/
polardbxengine
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
polardbxengine/storage/ndb/tools/NdbImportUtil.cpp

3983 lines
88 KiB
C++
Raw Permalink Blame History

/*
Copyright (c) 2018, 2019, Oracle and/or its affiliates. All rights reserved.
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License, version 2.0,
as published by the Free Software Foundation.
This program is also distributed with certain software (including
but not limited to OpenSSL) that is licensed under separate terms,
as designated in a particular file or component or in included license
documentation. The authors of MySQL hereby grant you an additional
permission to link the program and your derivative works with the
separately licensed software that they have included with MySQL.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License, version 2.0, for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include "NdbImportUtil.hpp"
#include "m_ctype.h"
#include "my_sys.h"
#include <NdbDictionaryImpl.hpp>
#include <Vector.hpp>
#include <inttypes.h>
// STL
#include <cmath>
NdbImportUtil::NdbImportUtil() :
m_util(*this),
c_stats(*this)
{
log_debug(1, "ctor");
c_rows_free = new RowList;
c_rows_free->set_stats(m_util.c_stats, "rows-free");
c_blobs_free = new BlobList;
add_pseudo_tables();
}
NdbImportUtil::~NdbImportUtil()
{
log_debug(1, "dtor");
delete c_blobs_free;
delete c_rows_free;
require(c_tables.m_tables.empty());
}
NdbOut&
operator<<(NdbOut& out, const NdbImportUtil& util)
{
out << "util ";
return out;
}
// DebugLogger
NdbImportUtil::DebugLogger::DebugLogger()
{
logfile = new FileOutputStream(stderr);
out = new NdbOut(* logfile);
mutex = NdbMutex_Create();
require(mutex != 0);
timer.start();
start.timer = & timer;
start.mutex = mutex;
stop.mutex = mutex;
}
NdbImportUtil::DebugLogger::~DebugLogger()
{
delete logfile;
delete out;
NdbMutex_Destroy(mutex);
}
NdbOut&
operator<<(NdbOut& out, const NdbImportUtil::DebugLogger::MessageStart & start)
{
/* Lock the logger mutex, obtain a timestamp, and print it. */
char buf[32];
NdbMutex_Lock(start.mutex);
start.timer->stop();
double t = (double) start.timer->elapsed_msec() / (double)1000.0;
sprintf(buf, "%8.3f: ", t);
out << buf;
return out;
}
NdbOut&
operator<<(NdbOut& out, const NdbImportUtil::DebugLogger::MessageStop & stop)
{
/* Print a final newline, then unlock the logger mutex. */
out << "\n";
NdbMutex_Unlock(stop.mutex);
return out;
}
// name
NdbImportUtil::Name::Name(const char* s)
{
require(s != 0);
m_str = s;
}
NdbImportUtil::Name::Name(const char* s, const char* t)
{
require(s != 0 && t != 0);
m_str = s;
m_str += "-";
m_str += t;
}
NdbImportUtil::Name::Name(const char* s, uint t)
{
require(s != 0);
m_str = s;
m_str += "-";
char b[100];
sprintf(b, "%u", t);
m_str += b;
}
NdbOut&
operator<<(NdbOut& out, const NdbImportUtil::Name& name)
{
out << (const char*)name;
return out;
}
// lockable
NdbImportUtil::Lockable::Lockable()
{
m_mutex = NdbMutex_Create();
m_condition = NdbCondition_Create();
require(m_mutex != 0 && m_condition != 0);
}
NdbImportUtil::Lockable::~Lockable()
{
NdbCondition_Destroy(m_condition);
NdbMutex_Destroy(m_mutex);
}
void
NdbImportUtil::Lockable::lock()
{
NdbMutex_Lock(m_mutex);
}
void
NdbImportUtil::Lockable::unlock()
{
NdbMutex_Unlock(m_mutex);
}
void
NdbImportUtil::Lockable::wait(uint timeout)
{
NdbCondition_WaitTimeout(m_condition, m_mutex, timeout);
}
void
NdbImportUtil::Lockable::signal()
{
NdbCondition_Signal(m_condition);
}
// thread
NdbImportUtil::Thread::Thread()
{
m_thread = 0;
}
NdbImportUtil::Thread::~Thread()
{
if (m_thread != 0)
NdbThread_Destroy(&m_thread);
}
void
NdbImportUtil::Thread::join()
{
require(m_thread != 0);
NdbThread_WaitFor(m_thread, (void**)0);
NdbThread_Destroy(&m_thread);
}
// list
NdbImportUtil::ListEnt::ListEnt()
{
m_next = 0;
m_prev = 0;
}
NdbImportUtil::ListEnt::~ListEnt()
{
}
NdbImportUtil::List::List()
{
m_front = 0;
m_back = 0;
m_cnt = 0;
m_maxcnt = 0;
m_totcnt = 0;
m_stat_occup = 0;
m_stat_total = 0;
}
NdbImportUtil::List::~List()
{
ListEnt* ent;
while ((ent = pop_front()) != 0)
delete ent;
}
void
NdbImportUtil::List::set_stats(Stats& stats, const char* name)
{
{
const Name statname(name, "occup");
Stat* stat = stats.create(statname, 0, 0);
m_stat_occup = stat;
}
{
const Name statname(name, "total");
Stat* stat = stats.create(statname, 0, 0);
m_stat_total = stat;
}
}
void
NdbImportUtil::List::push_back(ListEnt* ent)
{
require(ent != 0);
require(ent->m_next == 0 && ent->m_prev == 0);
if (m_cnt == 0)
{
m_front = m_back = ent;
ent->m_next = 0;
ent->m_prev = 0;
}
else
{
m_back->m_next = ent;
ent->m_next = 0;
ent->m_prev = m_back;
m_back = ent;
}
m_cnt++;
if (m_maxcnt < m_cnt)
{
m_maxcnt++;
require(m_maxcnt == m_cnt);
}
m_totcnt++;
validate();
if (m_stat_occup != 0)
m_stat_occup->add(m_cnt);
if (m_stat_total != 0)
m_stat_total->add(1);
}
void
NdbImportUtil::List::push_front(ListEnt* ent)
{
require(ent != 0);
require(ent->m_next == 0 && ent->m_prev == 0);
if (m_cnt == 0)
{
m_front = m_back = ent;
ent->m_next = 0;
ent->m_prev = 0;
}
else
{
m_front->m_prev = ent;
ent->m_prev = 0;
ent->m_next = m_front;
m_front = ent;
}
m_cnt++;
if (m_maxcnt < m_cnt)
{
m_maxcnt++;
require(m_maxcnt == m_cnt);
}
m_totcnt++;
validate();
if (m_stat_occup != 0)
m_stat_occup->add(m_cnt);
if (m_stat_total != 0)
m_stat_total->add(1);
}
void
NdbImportUtil::List::push_after(ListEnt* ent1, ListEnt* ent2)
{
require(ent1 != 0 && ent2 != 0);
require(ent2->m_next == 0 && ent2->m_prev == 0);
ListEnt* ent3 = ent1->m_next;
if (ent3 == 0)
{
push_back(ent2);
return;
}
ent1->m_next = ent2;
ent2->m_prev = ent1;
ent2->m_next = ent3;
ent3->m_prev = ent2;
m_cnt++;
if (m_maxcnt < m_cnt)
{
m_maxcnt++;
require(m_maxcnt == m_cnt);
}
m_totcnt++;
validate();
if (m_stat_occup != 0)
m_stat_occup->add(m_cnt);
if (m_stat_total != 0)
m_stat_total->add(1);
}
void
NdbImportUtil::List::push_before(ListEnt* ent1, ListEnt* ent2)
{
require(ent1 != 0 && ent2 != 0);
require(ent2->m_next == 0 && ent2->m_prev == 0);
ListEnt* ent3 = ent1->m_prev;
if (ent3 == 0)
{
push_front(ent2);
return;
}
ent1->m_prev = ent2;
ent2->m_next = ent1;
ent2->m_prev = ent3;
ent3->m_next = ent2;
m_cnt++;
if (m_maxcnt < m_cnt)
{
m_maxcnt++;
require(m_maxcnt == m_cnt);
}
m_totcnt++;
validate();
if (m_stat_occup != 0)
m_stat_occup->add(m_cnt);
if (m_stat_total != 0)
m_stat_total->add(1);
}
NdbImportUtil::ListEnt*
NdbImportUtil::List::pop_front()
{
ListEnt* ent = 0;
if (m_cnt != 0)
{
if (m_cnt == 1)
{
ent = m_front;
m_front = m_back = 0;
}
else
{
ent = m_front;
m_front = ent->m_next;
m_front->m_prev = 0;
ent->m_next = 0;
ent->m_prev = 0;
}
m_cnt--;
validate();
if (m_stat_occup != 0)
m_stat_occup->add(m_cnt);
}
return ent;
}
void
NdbImportUtil::List::remove(ListEnt* ent)
{
ListEnt* prev = ent->m_prev;
ListEnt* next = ent->m_next;
ent->m_prev = 0;
ent->m_next = 0;
if (prev != 0)
prev->m_next = next;
if (next != 0)
next->m_prev = prev;
if (m_front == ent)
m_front = next;
if (m_back == ent)
m_back = prev;
require(m_cnt != 0);
m_cnt--;
validate();
if (m_stat_occup != 0)
m_stat_occup->add(m_cnt);
}
void
NdbImportUtil::List::push_back_from(List& src)
{
if (src.m_cnt != 0)
{
if (m_cnt != 0)
{
ListEnt* ent1 = m_back;
ListEnt* ent2 = src.m_front;
require(ent1 != 0 && ent2 != 0);
require(ent1->m_next == 0 && ent2->m_prev == 0);
// push src to the back
ent1->m_next = ent2;
ent2->m_prev = ent1;
m_back = src.m_back;
m_cnt += src.m_cnt;
}
else
{
m_front = src.m_front;
m_back = src.m_back;
m_cnt = src.m_cnt;
}
if (m_maxcnt < m_cnt)
m_maxcnt = m_cnt;
m_totcnt += src.m_cnt;
}
validate();
// erase src but leave stats alone
src.m_front = 0;
src.m_back = 0;
src.m_cnt = 0;
}
#if defined(VM_TRACE) || defined(TEST_NDBIMPORTUTIL)
void
NdbImportUtil::List::validate() const
{
if (m_cnt == 0)
{
require(m_front == 0);
require(m_back == 0);
}
else
{
require(m_front != 0);
require(m_front->m_prev == 0);
require(m_back != 0);
require(m_back->m_next == 0);
if (m_cnt == 1)
require(m_front == m_back);
else
require(m_front != m_back);
}
#if defined(VM_TRACE) && defined(TEST_NDBIMPORTUTIL)
uint cnt = 0;
ListEnt* ent1 = m_front;
ListEnt* ent2 = 0;
while (ent1 != 0)
{
require(ent1->m_prev == ent2);
if (ent2 != 0)
require(ent2->m_next == ent1);
ent2 = ent1;
ent1 = ent1->m_next;
cnt++;
}
require(m_cnt == cnt);
#endif
}
#endif
// attrs
NdbImportUtil::Attr::Attr()
{
m_attrname = "";
m_attrno = Inval_uint;
m_attrid = Inval_uint;
m_type = NdbDictionary::Column::Undefined;
m_charset = 0;
memset(m_sqltype, 0, sizeof(m_sqltype));
m_pk = false;
m_nullable = false;
m_precision = 0;
m_scale = 0;
m_length = 0;
m_charlength = 0;
m_arraytype = NdbDictionary::Column::ArrayTypeFixed;
m_inlinesize = 0;
m_partsize = 0;
m_blobtable = 0;
m_size = 0;
m_pad = false;
m_padchar = 0;
m_quotable = false;
m_isblob = false;
m_blobno = Inval_uint;
m_offset = 0;
m_null_byte = Inval_uint;
m_null_bit = Inval_uint;
}
void
NdbImportUtil::Attr::set_value(Row* row, const void* data, uint len) const
{
require(!m_isblob);
require(data != 0);
uint totlen = m_arraytype + len;
require(totlen <= m_size);
require(m_offset + totlen <= row->m_recsize);
uchar* p = &row->m_data[m_offset];
switch (m_arraytype) {
case 0:
break;
case 1:
require(len <= 0xff);
p[0] = (uchar)len;
p += 1;
break;
case 2:
require(len <= 0xffff);
p[0] = (uchar)(len & 0xff);
p[1] = (uchar)(len >> 8);
p += 2;
break;
default:
require(false);
break;
}
memcpy(p, data, len);
if (m_pad)
memset(&p[totlen], m_padchar, m_size - totlen);
if (m_nullable)
set_null(row, false);
}
void
NdbImportUtil::Attr::set_blob(Row* row, const void* data, uint len) const
{
require(m_isblob);
require(data != 0);
require(m_blobno < row->m_blobs.size());
Blob* blob = row->m_blobs[m_blobno];
require(blob != 0);
blob->resize(len);
memcpy(blob->m_data, data, len);
blob->m_blobsize = len;
if (m_nullable)
set_null(row, false);
// add to rowsize which already includes recsize
require(row->m_rowsize >= row->m_recsize);
row->m_rowsize += len;
}
void
NdbImportUtil::Attr::set_null(Row* row, bool null) const
{
uchar* data = row->m_data;
uchar mask = (1 << m_null_bit);
if (null)
data[m_null_byte] |= mask;
else
data[m_null_byte] &= ~mask;
}
const uchar*
NdbImportUtil::Attr::get_value(const Row* row) const
{
const uchar* p = &row->m_data[m_offset];
return p;
}
void
NdbImportUtil::Attr::get_value(const Row* row, uint32& value) const
{
const uchar* p = get_value(row);
if (m_type == NdbDictionary::Column::Unsigned)
{
memcpy(&value, p, sizeof(value));
return;
}
require(false);
}
void
NdbImportUtil::Attr::get_value(const Row* row, uint64& value) const
{
const uchar* p = get_value(row);
if (m_type == NdbDictionary::Column::Bigunsigned)
{
memcpy(&value, p, sizeof(value));
return;
}
require(false);
}
void
NdbImportUtil::Attr::get_value(const Row* row, char* buf, uint bufsz) const
{
const uchar* p = get_value(row);
memset(buf, 0, bufsz);
if (m_type == NdbDictionary::Column::Varchar) {
uint sz = p[0];
require(sz < bufsz);
memcpy(buf, &p[1], sz);
return;
}
if (m_type == NdbDictionary::Column::Longvarchar) {
uint sz = p[0] | (p[1] << 8);
require(sz < bufsz);
memcpy(buf, &p[2], sz);
return;
}
require(false);
}
bool
NdbImportUtil::Attr::get_null(const Row* row) const
{
bool null = false;
if (m_nullable)
{
const uchar* data = row->m_data;
uchar mask = (1 << m_null_bit);
null = (bool)(data[m_null_byte] & mask);
}
return null;
}
uint
NdbImportUtil::Attr::get_blob_parts(uint len) const
{
require(m_isblob);
uint parts = 0;
if (len > m_inlinesize)
{
require(m_partsize != 0);
parts = (len -m_inlinesize + m_partsize - 1) / m_partsize;
}
return parts;
}
// could go in NdbDictionary
void
NdbImportUtil::Attr::set_sqltype()
{
m_sqltype[0] = 0;
switch (m_type) {
case NdbDictionary::Column::Tinyint:
sprintf(m_sqltype, "tinyint");
break;
case NdbDictionary::Column::Smallint:
sprintf(m_sqltype, "smallint");
break;
case NdbDictionary::Column::Mediumint:
sprintf(m_sqltype, "mediumint");
break;
case NdbDictionary::Column::Int:
sprintf(m_sqltype, "int");
break;
case NdbDictionary::Column::Bigint:
sprintf(m_sqltype, "bigint");
break;
case NdbDictionary::Column::Tinyunsigned:
sprintf(m_sqltype, "tinyint unsigned");
break;
case NdbDictionary::Column::Smallunsigned:
sprintf(m_sqltype, "smallint unsigned");
break;
case NdbDictionary::Column::Mediumunsigned:
sprintf(m_sqltype, "mediumint unsigned");
break;
case NdbDictionary::Column::Unsigned:
sprintf(m_sqltype, "int unsigned");
break;
case NdbDictionary::Column::Bigunsigned:
sprintf(m_sqltype, "bigint unsigned");
break;
case NdbDictionary::Column::Decimal:
sprintf(m_sqltype, "decimal");
break;
case NdbDictionary::Column::Decimalunsigned:
sprintf(m_sqltype, "decimal unsigned");
break;
case NdbDictionary::Column::Float:
sprintf(m_sqltype, "float");
break;
case NdbDictionary::Column::Double:
sprintf(m_sqltype, "double");
break;
case NdbDictionary::Column::Char:
require(m_charset != 0 && m_charset->csname != 0);
sprintf(m_sqltype, "char(%u) %s", m_charlength, m_charset->csname);
break;
case NdbDictionary::Column::Varchar:
require(m_charset != 0 && m_charset->csname != 0);
sprintf(m_sqltype, "varchar(%u) %s", m_charlength, m_charset->csname);
break;
case NdbDictionary::Column::Longvarchar:
require(m_charset != 0 && m_charset->csname != 0);
sprintf(m_sqltype, "varchar(%u) %s", m_charlength, m_charset->csname);
break;
case NdbDictionary::Column::Binary:
sprintf(m_sqltype, "binary(%u)", m_length);
break;
case NdbDictionary::Column::Varbinary:
sprintf(m_sqltype, "varbinary(%u)", m_length);
break;
case NdbDictionary::Column::Longvarbinary:
sprintf(m_sqltype, "varbinary(%u)", m_length);
break;
case NdbDictionary::Column::Bit:
sprintf(m_sqltype, "bit(%u)", m_length);
break;
case NdbDictionary::Column::Year:
sprintf(m_sqltype, "year");
break;
case NdbDictionary::Column::Date:
sprintf(m_sqltype, "date");
break;
case NdbDictionary::Column::Time2:
if (m_precision == 0)
sprintf(m_sqltype, "time");
else
sprintf(m_sqltype, "time(%u)", m_precision);
break;
case NdbDictionary::Column::Datetime2:
if (m_precision == 0)
sprintf(m_sqltype, "datetime");
else
sprintf(m_sqltype, "datetime(%u)", m_precision);
break;
case NdbDictionary::Column::Timestamp2:
if (m_precision == 0)
sprintf(m_sqltype, "timestamp");
else
sprintf(m_sqltype, "timestamp(%u)", m_precision);
break;
case NdbDictionary::Column::Blob:
sprintf(m_sqltype, "blob");
break;
case NdbDictionary::Column::Text:
sprintf(m_sqltype, "text");
break;
default:
sprintf(m_sqltype, "unknown type=%d", (int)m_type);
break;
}
}
// tables
NdbImportUtil::Table::Table()
{
m_tabid = Inval_uint;
m_tab = 0;
m_rec = 0;
m_keyrec = NULL;
m_recsize = 0;
m_has_hidden_pk = false;
}
void
NdbImportUtil::Table::add_pseudo_attr(const char* name,
NdbDictionary::Column::Type type,
uint length)
{
const uint id = m_attrs.size();
Attr attr; // ctor sets defaults
attr.m_attrname = name;
attr.m_attrno = id;
attr.m_attrid = id;
attr.m_type = type;
attr.m_length = length;
attr.m_charlength = length;
switch (type) {
case NdbDictionary::Column::Unsigned:
require(length == 1);
attr.m_arraytype = NdbDictionary::Column::ArrayTypeFixed;
attr.m_size = 4;
attr.m_quotable = false;
break;
case NdbDictionary::Column::Bigunsigned:
require(length == 1);
attr.m_arraytype = NdbDictionary::Column::ArrayTypeFixed;
attr.m_size = 8;
attr.m_quotable = false;
break;
case NdbDictionary::Column::Double:
require(length == 1);
attr.m_arraytype = NdbDictionary::Column::ArrayTypeFixed;
attr.m_size = 8;
attr.m_quotable = false;
break;
case NdbDictionary::Column::Varchar:
require(length > 1);
attr.m_charset = &my_charset_bin;
attr.m_arraytype = NdbDictionary::Column::ArrayTypeShortVar;
attr.m_size = 1 + length;
attr.m_quotable = true;
break;
case NdbDictionary::Column::Longvarchar:
require(length > 1);
attr.m_charset = &my_charset_bin;
attr.m_arraytype = NdbDictionary::Column::ArrayTypeMediumVar;
attr.m_size = 2 + length;
attr.m_quotable = true;
break;
case NdbDictionary::Column::Text:
attr.m_charset = &my_charset_bin;
attr.m_inlinesize = 256;
attr.m_partsize = 2000;
attr.m_isblob = true;
attr.m_blobno = m_blobids.size();
m_blobids.push_back(id);
attr.m_quotable = true;
break;
default:
require(false);
break;
};
attr.set_sqltype();
if (id == 0)
attr.m_offset = 0;
else
{
const Attr& prevattr = m_attrs[id - 1];
attr.m_offset = prevattr.m_offset + prevattr.m_size;
}
attr.m_null_byte = Inval_uint;
attr.m_null_bit = Inval_uint;
m_recsize += attr.m_size;
m_attrs.push_back(attr);
}
const NdbImportUtil::Attr&
NdbImportUtil::Table::get_attr(const char* attrname) const
{
uint i = 0;
const uint n = m_attrs.size();
while (i < n)
{
if (strcmp(m_attrs[i].m_attrname.c_str(), attrname) == 0)
break;
i++;
}
require(i < n);
return m_attrs[i];
}
uint
NdbImportUtil::Table::get_nodeid(uint fragid) const
{
require(fragid < m_fragments.size());
uint nodeid = m_fragments[fragid];
return nodeid;
}
int
NdbImportUtil::add_table(NdbDictionary::Dictionary* dic,
const NdbDictionary::Table* tab,
uint& tabid,
Error& error)
{
require(tab != 0);
require(tab->getObjectStatus() == NdbDictionary::Object::Retrieved);
log_debug(1, "add_table: " << tab->getName());
tabid = tab->getObjectId();
// check if mapped already
{
std::map<uint, Table>::const_iterator it;
it = c_tables.m_tables.find(tabid);
if (it != c_tables.m_tables.end())
{
const Table& table = it->second;
require(table.m_tab == tab);
return 0;
}
}
Table table;
do
{
const NdbRecord* rec = tab->getDefaultRecord();
table.m_tabid = tabid;
table.m_tab = tab;
table.m_rec = rec;
table.m_recsize = NdbDictionary::getRecordRowLength(rec);
Attrs& attrs = table.m_attrs;
const uint attrcnt = tab->getNoOfColumns();
attrs.reserve(attrcnt);
bool ok = true;
Uint32 recAttrId;
for (uint i = 0; i < attrcnt && ok; i++)
{
if (i == 0)
require(NdbDictionary::getFirstAttrId(rec, recAttrId));
else
require(NdbDictionary::getNextAttrId(rec, recAttrId));
require(recAttrId == i);
Attr attr;
const NdbDictionary::Column* col = tab->getColumn(i);
require(col !=0);
attr.m_attrname = col->getName();
attr.m_attrno = i;
attr.m_attrid = i;
attr.m_type = col->getType();
attr.m_pk = col->getPrimaryKey();
attr.m_nullable = col->getNullable();
attr.m_precision = col->getPrecision();
attr.m_scale = col->getScale();
attr.m_length = col->getLength();
attr.m_arraytype = col->getArrayType();
require(attr.m_arraytype <= 2);
attr.m_size = col->getSizeInBytes();
switch (attr.m_type) {
case NdbDictionary::Column::Char:
attr.m_pad = true;
attr.m_padchar = 0x20;
break;
case NdbDictionary::Column::Binary:
attr.m_pad = true;
attr.m_padchar = 0x0;
break;
default:
attr.m_pad = false;
break;
}
switch (attr.m_type) {
case NdbDictionary::Column::Char:
case NdbDictionary::Column::Varchar:
case NdbDictionary::Column::Longvarchar:
attr.m_charset = col->getCharset();
require(attr.m_charset != 0);
uint mbmaxlen; mbmaxlen = attr.m_charset->mbmaxlen;
require(mbmaxlen != 0);
require(attr.m_length % mbmaxlen == 0);
attr.m_charlength = attr.m_length / mbmaxlen;
attr.m_quotable = true;
break;
case NdbDictionary::Column::Text:
attr.m_charset = col->getCharset();
require(attr.m_charset != 0);
break;
case NdbDictionary::Column::Binary:
case NdbDictionary::Column::Varbinary:
case NdbDictionary::Column::Longvarbinary:
attr.m_charset = 0;
attr.m_charlength = attr.m_length;
attr.m_quotable = true;
break;
default:
attr.m_charset = 0;
attr.m_charlength = attr.m_length;
attr.m_quotable = false;
break;
}
switch (attr.m_type) {
case NdbDictionary::Column::Blob:
case NdbDictionary::Column::Text:
attr.m_isblob = true;
attr.m_inlinesize = col->getInlineSize();
attr.m_partsize = col->getPartSize();
attr.m_blobno = table.m_blobids.size();
attr.m_blobtable = col->getBlobTable();
if (attr.m_partsize == 0)
require(attr.m_blobtable == 0);
else
require(attr.m_blobtable != 0);
table.m_blobids.push_back(i);
break;
default:
attr.m_isblob = false;
break;
}
attr.set_sqltype();
Uint32 offset;
require(NdbDictionary::getOffset(rec, i, offset));
attr.m_offset = offset;
Uint32 null_byte, null_bit;
require(NdbDictionary::getNullBitOffset(rec, i, null_byte, null_bit));
attr.m_null_byte = null_byte;
attr.m_null_bit = null_bit;
attrs.push_back(attr);
}
if (!ok)
break;
require(!NdbDictionary::getNextAttrId(rec, recAttrId));
NdbDictionary::RecordSpecification
speclist[NDB_MAX_NO_OF_ATTRIBUTES_IN_KEY];
uint nkey = 0;
for (uint i = 0; i < attrcnt && ok; i++)
{
const Attr& attr = attrs[i];
if (attr.m_pk)
{
const NdbDictionary::Column* col = tab->getColumn(i);
require(col->getPrimaryKey());
NdbDictionary::RecordSpecification& spec = speclist[nkey];
spec.column = col;
spec.offset = attr.m_offset;
spec.nullbit_byte_offset = attr.m_null_byte;
spec.nullbit_bit_in_byte = attr.m_null_bit;
nkey++;
// guess hidden pk
if (strcmp(attr.m_attrname.c_str(), "$PK") == 0)
{
if (i + 1 == attrcnt &&
nkey == 1 &&
attr.m_type == NdbDictionary::Column::Bigunsigned)
table.m_has_hidden_pk = true;
else
{
set_error_usage(error, __LINE__,
"column %u: "
"invalid use of reserved column name $PK", i);
ok = false;
break;
}
}
}
}
if (!ok)
break;
require(nkey == (uint)tab->getNoOfPrimaryKeys());
const NdbRecord* keyrec =
dic->createRecord(tab,
speclist,
nkey,
sizeof(NdbDictionary::RecordSpecification));
if (keyrec == 0)
{
m_util.set_error_ndb(error, __LINE__, dic->getNdbError());
break;
}
table.m_keyrec = keyrec;
{
NdbTableImpl& tabImpl = NdbTableImpl::getImpl(*tab);
const Vector<Uint16>& fragments = tabImpl.m_fragments;
for (uint i = 0; i < fragments.size(); i++)
{
uint16 nodeid = fragments[i];
table.m_fragments.push_back(nodeid);
}
}
c_tables.m_tables.insert(std::pair<uint, Table>(tabid, table));
return 0;
} while (0);
return -1;
}
const NdbImportUtil::Table&
NdbImportUtil::get_table(uint tabid)
{
std::map<uint, Table>::const_iterator it;
it = c_tables.m_tables.find(tabid);
require(it != c_tables.m_tables.end());
const Table& table = it->second;
return table;
}
void
NdbImportUtil::remove_table(NdbDictionary::Dictionary* dic, uint tabid)
{
std::map<uint, Table>::const_iterator it;
it = c_tables.m_tables.find(tabid);
require(it != c_tables.m_tables.end());
const Table& table = it->second;
dic->releaseRecord(const_cast<NdbRecord*>(table.m_keyrec));
c_tables.m_tables.erase(it);
}
// rows
NdbImportUtil::Row::Row()
{
m_tabid = Inval_uint;
m_recsize = 0;
m_rowsize = 0;
m_allocsize = 0;
m_rowid = Inval_uint64;
m_linenr = Inval_uint64;
m_startpos = Inval_uint64;
m_endpos = Inval_uint64;
m_data = 0;
}
NdbImportUtil::Row::~Row()
{
delete [] m_data;
for (uint i = 0; i < m_blobs.size(); ++i)
{
Blob* blob = m_blobs[i];
if (blob != NULL)
{
delete blob;
}
}
m_blobs.clear();
}
void
NdbImportUtil::Row::init(const Table& table)
{
m_tabid = table.m_tabid;
uint recsize = table.m_recsize;
require(recsize > 0);
m_recsize = recsize;
m_rowsize = recsize; // full main record is always included
if (m_allocsize < recsize)
{
delete [] m_data;
m_data = new uchar [recsize];
m_allocsize = recsize;
}
}
NdbImportUtil::RowList::RowList()
{
m_rowsize = 0;
m_rowbatch = UINT_MAX;
m_rowbytes = UINT_MAX;
m_eof = false;
m_foe = false;
m_overflow = 0;
m_underflow = 0;
m_stat_overflow = 0;
m_stat_underflow = 0;
m_stat_locks = 0;
}
NdbImportUtil::RowList::~RowList ()
{
Row *one_row;
while ((one_row = pop_front()) != NULL)
{
delete one_row;
}
}
void
NdbImportUtil::RowList::set_stats(Stats& stats, const char* name)
{
List::set_stats(stats, name);
{
const Name statname(name, "overflow");
Stat* stat = stats.create(statname, 0, 0);
m_stat_overflow = stat;
}
{
const Name statname(name, "underflow");
Stat* stat = stats.create(statname, 0, 0);
m_stat_underflow = stat;
}
{
const Name statname(name, "locks");
Stat* stat = stats.create(statname, 0, 0);
m_stat_locks = stat;
}
}
bool
NdbImportUtil::RowList::push_back(Row* row)
{
bool ret = false;
if (m_cnt < m_rowbatch && m_rowsize < m_rowbytes)
{
List::push_back(row);
m_rowsize += row->m_rowsize;
ret = true;
}
else
{
m_overflow++;
if (m_stat_overflow != 0)
m_stat_overflow->add(1);
}
return ret;
}
void
NdbImportUtil::RowList::push_back_force(Row* row)
{
List::push_back(row);
m_rowsize += row->m_rowsize;
}
bool
NdbImportUtil::RowList::push_front(Row* row)
{
bool ret = false;
if (m_cnt < m_rowbatch && m_rowsize < m_rowbytes)
{
List::push_front(row);
m_rowsize += row->m_rowsize;
ret = true;
}
else
{
m_overflow++;
if (m_stat_overflow != 0)
m_stat_overflow->add(1);
}
return ret;
}
void
NdbImportUtil::RowList::push_front_force(Row* row)
{
List::push_front(row);
m_rowsize += row->m_rowsize;
}
NdbImportUtil::Row*
NdbImportUtil::RowList::pop_front()
{
Row* row = 0;
do
{
row = static_cast<Row*>(List::pop_front());
if (row != 0)
{
require(m_rowsize >= row->m_rowsize);
m_rowsize -= row->m_rowsize;
break;
}
m_underflow++;
if (m_stat_underflow != 0)
m_stat_underflow->add(1);
} while (0);
return row;
}
void
NdbImportUtil::RowList::remove(Row* row)
{
List::remove(row);
require(m_rowsize >= row->m_rowsize);
m_rowsize -= row->m_rowsize;
}
void
NdbImportUtil::RowList::push_back_from(RowList& src)
{
List::push_back_from(src);
m_rowsize += src.m_rowsize;
src.m_rowsize = 0;
validate();
src.validate();
}
/*
* Transfer rows from a shared list src to our list. If src is
* empty, try to wait. Terminate if our list is full. If any rows
* were transferred, do not wait for more, and signal that src now
* has fewer rows.
*/
void
NdbImportUtil::RowList::push_back_from(RowList& src, RowCtl& ctl)
{
uint retries = ctl.m_retries;
uint cnt_out = 0;
uint bytes_out = 0;
if (unlikely(full()))
return;
while (src.empty() && retries != 0)
{
if (ctl.m_dowait)
src.wait(ctl.m_timeout);
retries--;
}
while (!src.empty())
{
// pop because row cannot be on 2 lists
Row* row = src.pop_front();
if (push_back(row))
{
cnt_out++;
bytes_out += row->m_rowsize;
continue;
}
src.push_front_force(row);
// our list is full
break;
}
if (cnt_out != 0 && ctl.m_dosignal)
{
// signal that we removed some rows from src
src.signal();
}
ctl.m_cnt_out += cnt_out;
ctl.m_bytes_out += bytes_out;
}
/*
* Transfer rows from our list to a shared list dst. If dst is
* full, try to wait. Terminate if our list is empty. If any rows
* were transferred, do not wait for more, and signal that dst now
* has more rows.
*/
void
NdbImportUtil::RowList::pop_front_to(RowList& dst, RowCtl& ctl)
{
uint retries = ctl.m_retries;
uint cnt_out = 0;
uint bytes_out = 0;
if (unlikely(empty()))
return;
while (dst.full() && retries != 0)
{
if (ctl.m_dowait)
dst.wait(ctl.m_timeout);
retries--;
}
while (!empty())
{
// pop because row cannot be on 2 lists
Row* row = pop_front();
if (dst.push_back(row))
{
cnt_out++;
bytes_out += row->m_rowsize;
continue;
}
push_front_force(row);
// dst is full
break;
}
if (cnt_out != 0 && ctl.m_dosignal)
{
// signal that we added some rows to dst
dst.signal();
}
ctl.m_cnt_out += cnt_out;
ctl.m_bytes_out += bytes_out;
}
#if defined(VM_TRACE) || defined(TEST_NDBIMPORTUTIL)
void
NdbImportUtil::RowList::validate() const
{
List::validate();
if (m_cnt == 0)
require(m_rowsize == 0);
if (m_rowsize == 0)
require(m_cnt == 0);
#if defined(VM_TRACE) && defined(TEST_NDBIMPORTUTIL)
uint rowsize = 0;
const Row* row = static_cast<const Row*>(m_front);
while (row != 0)
{
rowsize += row->m_rowsize;
row = static_cast<const Row*>(row->m_next);
}
require(m_rowsize == rowsize);
#endif
}
#endif
// alloc and free shared rows
NdbImportUtil::Row*
NdbImportUtil::alloc_row(const Table& table, bool dolock)
{
RowList& rows = *c_rows_free;
if (dolock)
rows.lock();
Row* row = rows.pop_front();
if (dolock)
rows.unlock();
if (row == 0)
{
row = new Row;
}
row->init(table);
while (row->m_blobs.size() < table.m_blobids.size())
{
Blob* blob = alloc_blob();
row->m_blobs.push_back(blob);
}
return row;
}
void
NdbImportUtil::alloc_rows(const Table& table, uint cnt, RowList& dst)
{
RowList& rows = *c_rows_free;
rows.lock();
for (uint i = 0; i < cnt; i++)
{
Row* row = alloc_row(table, false);
dst.push_back_force(row); // ignore limits
}
rows.unlock();
}
void
NdbImportUtil::free_blobs_from_row(Row *row)
{
for (uint i = 0; i < row->m_blobs.size(); ++i)
{
Blob* blob = row->m_blobs[i];
if (blob != NULL)
{
free_blob(blob);
}
}
row->m_blobs.clear();
}
void
NdbImportUtil::free_row(Row* row)
{
free_blobs_from_row(row);
RowList& rows = *c_rows_free;
rows.lock();
rows.push_back(row);
rows.unlock();
}
void
NdbImportUtil::free_rows(RowList& src)
{
RowList blob_freed_rows;
while (!src.empty())
{
Row *one_row = src.pop_front();
free_blobs_from_row(one_row);
blob_freed_rows.push_back(one_row);
}
RowList& rows = *c_rows_free;
rows.lock();
rows.push_back_from(blob_freed_rows);
rows.unlock();
}
// blobs
NdbImportUtil::Blob::Blob()
{
m_blobsize = 0;
m_allocsize = 0;
m_data = new uchar [0];
}
NdbImportUtil::Blob::~Blob()
{
delete [] m_data;
}
NdbImportUtil::BlobList::BlobList()
{
}
NdbImportUtil::BlobList::~BlobList()
{
Blob *blob = NULL;
while ((blob = pop_front()) != NULL)
{
delete blob;
}
}
void
NdbImportUtil::Blob::resize(uint size)
{
if (m_allocsize < size)
{
delete [] m_data;
m_data = new uchar [size];
m_allocsize = size;
}
}
NdbImportUtil::Blob*
NdbImportUtil::alloc_blob()
{
BlobList& blobs = *c_blobs_free;
blobs.lock();
Blob* blob = static_cast<Blob*>(blobs.pop_front());
blobs.unlock();
if (blob == 0)
{
blob = new Blob;
}
return blob;
}
void
NdbImportUtil::free_blob(Blob* blob)
{
BlobList& blobs = *c_blobs_free;
blobs.lock();
blobs.push_back(blob);
blobs.unlock();
}
// rowmap
NdbImportUtil::RowMap::RowMap(NdbImportUtil& util) :
m_util(util)
{
}
NdbImportUtil::Range::Range()
{
m_start = 0;
m_end = 0;
m_startpos = 0;
m_endpos = 0;
m_reject = 0;
}
NdbImportUtil::Range::~Range()
{
}
void
NdbImportUtil::Range::copy(const Range& range2)
{
m_start = range2.m_start;
m_end = range2.m_end;
m_startpos = range2.m_startpos;
m_endpos = range2.m_endpos;
m_reject = range2.m_reject;
}
/*
* This is typically used by a worker to add a row to its
* private rowmap. The row is likely to go near the end
* so search is done backwards.
*/
void
NdbImportUtil::RowMap::add(Range range2)
{
RangeList& ranges = m_ranges;
Range* r2 = alloc_range();
r2->copy(range2);
if (ranges.empty())
{
ranges.push_back(r2);
}
else
{
Range* rback = ranges.back();
Range* rfront = ranges.front();
if (rback->m_start < r2->m_start)
{
if (merge_up(rback, r2))
{
// rback grows up to include r2
free_range(r2);
}
else
{
ranges.push_back(r2);
}
}
else if (r2->m_start < rfront->m_start)
{
if (merge_down(rfront, r2))
{
// rfront grows down to include r2
free_range(r2);
}
else
{
ranges.push_front(r2);
}
}
else
{
// r2 is between 2 entries rprev rnext
Range* rprev = rback;
Range* rnext = 0;
while (1)
{
if (r2->m_start > rprev->m_start)
{
// found the place
require(rnext != 0);
if (merge_up(rprev, r2))
{
// rprev grows up to include r2
free_range(r2);
if (merge_up(rprev, rnext))
{
// rprev and rnext have been joined via r2
// rnext is now obsolete
ranges.remove(rnext);
free_range(rnext);
}
}
else if (merge_down(rnext, r2))
{
// rnext grows down to include r2
free_range(r2);
}
else
{
// r2 becomes new entry after rprev
ranges.push_after(rprev, r2);
}
break;
}
rnext = rprev;
rprev = rprev->prev();
require(rprev != 0);
}
}
}
validate();
}
/*
* Merge from another rowmap. Walks through both maps
* in ascending order. The argument map2 is not modified
* here but is normally cleared afterwards by caller.
*/
void
NdbImportUtil::RowMap::add(const RowMap& map2)
{
RangeList& ranges = m_ranges;
const RangeList& ranges2 = map2.m_ranges;
Range* r = ranges.front();
const Range* r2 = ranges2.front();
while (1)
{
if (r == 0)
{
// copy rest of map2 using our free ranges
while (r2 != 0)
{
r = alloc_range();
r->copy(*r2);
ranges.push_back(r);
r2 = r2->next();
}
break;
}
if (r2 == 0)
{
// nothing more to do
break;
}
if (r->m_start < r2->m_start)
{
{
Range* rnext = r->next();
if (rnext != 0 && rnext->m_start < r2->m_start)
{
// still below r2
r = rnext;
continue;
}
}
if (merge_up(r, r2))
{
// r grows up to include r2
Range* rnext = r->next();
if (rnext != 0 && merge_up(r, rnext))
{
// r and rnext have been joined via r2
// rnext is now obsolete
ranges.remove(rnext);
free_range(rnext);
}
// leave r unchanged as next r2 may also apply
{
const Range* r2next = r2->next();
// even in the join case r2next cannot overlap r
if (r2next != 0)
require(r->m_end <= r2next->m_start);
}
}
else
{
// r2 creates new entry
Range* rnext = alloc_range();
rnext->copy(*r2);
ranges.push_after(r, rnext);
require(rnext == r->next());
// move to the new entry
r = r->next();
{
Range* rnext = r->next();
if (rnext != 0 && merge_up(r, rnext))
{
// r and rnext have been joined via r2
// rnext is now obsolete
ranges.remove(rnext);
free_range(rnext);
}
}
// leave current r unchanged
}
// r2 has been consumed
r2 = r2->next();
continue;
}
if (r->m_start > r2->m_start)
{
if (merge_down(r, r2))
{
// r grows down to include r2
// no more entries below r but there can be one above
}
else
{
// r2 creates new entry
Range* rprev = alloc_range();
rprev->copy(*r2);
ranges.push_before(r, rprev);
// can be more entries below r
}
// r2 has been consumed
r2 = r2->next();
continue;
}
require(false);
}
validate();
}
/*
* find() and remove() are used only on --resume, which consumes
* the old rowmap. They need not be afficient.
*/
NdbImportUtil::Range*
NdbImportUtil::RowMap::find(uint64 rowid)
{
RangeList& ranges = m_ranges;
Range* r = ranges.front();
while (r != 0)
{
if (r->m_start <= rowid && rowid < r->m_end)
break;
r = r->next();
}
return r;
}
bool
NdbImportUtil::RowMap::remove(uint64 rowid)
{
RangeList& ranges = m_ranges;
Range* r = find(rowid);
if (r == 0)
return false;
if (rowid == r->m_start)
{
r->m_start++;
if (r->m_start == r->m_end)
{
ranges.remove(r);
free_range(r);
}
}
else if (rowid == r->m_end - 1)
{
r->m_end -= 1;
require(r->m_start < r->m_end);
}
else
{
Range* r2 = alloc_range();
r2->m_start = rowid + 1;
r2->m_end = r->m_end;
r2->m_reject = 0; // not relevant
require(r2->m_start < r2->m_end);
r->m_end = rowid;
require(r->m_start < r->m_end);
ranges.push_after(r, r2);
}
return true;
}
void
NdbImportUtil::RowMap::get_total(uint64& rows, uint64& reject) const
{
uint64 trows = 0;
uint64 treject = 0;
const Range* r = m_ranges.front();
while (r != 0)
{
trows += r->m_end - r->m_start - r->m_reject;
treject += r->m_reject;
r = r->next();
}
rows = trows;
reject = treject;
}
NdbImportUtil::Range*
NdbImportUtil::alloc_range(bool dolock)
{
RangeList& ranges = c_ranges_free;
if (dolock)
ranges.lock();
Range* range = ranges.pop_front();
if (dolock)
ranges.unlock();
if (range == 0) {
range = new Range;
}
return range;
}
void
NdbImportUtil::alloc_ranges(uint cnt, RangeList& dst)
{
RangeList& ranges = c_ranges_free;
ranges.lock();
for (uint i = 0; i < cnt; i++)
{
Range* range = alloc_range(false);
dst.push_back(range);
}
ranges.unlock();
}
void
NdbImportUtil::free_range(Range* range)
{
RangeList& ranges = c_ranges_free;
ranges.lock();
ranges.push_back(range);
ranges.unlock();
}
void
NdbImportUtil::free_ranges(RangeList& src)
{
RangeList& ranges = c_ranges_free;
ranges.lock();
ranges.push_back_from(src);
ranges.unlock();
}
#if defined(VM_TRACE) || defined(TEST_NDBIMPORTUTIL)
void
NdbImportUtil::RowMap::validate() const
{
m_ranges.validate();
#if defined(VM_TRACE) && defined(TEST_NDBIMPORTUTIL)
const RangeList& ranges = m_ranges;
const Range* r2 = 0;
const Range* r1 = ranges.front();
while (r1 != 0)
{
require(r1->m_start < r1->m_end);
if (r2 != 0)
require(r2->m_end < r1->m_start);
r2 = r1;
r1 = r1->next();
}
#endif
}
#endif
NdbOut&
operator<<(NdbOut& out, const NdbImportUtil::Range& range)
{
out << "start=" << range.m_start
<< " end=" << range.m_end
<< " rows=" << range.m_end - range.m_start
<< " startpos=" << range.m_startpos
<< " endpos=" << range.m_endpos
<< " bytes=" << range.m_endpos - range.m_startpos
<< " reject=" << range.m_reject;
return out;
}
NdbOut&
operator<<(NdbOut& out, const NdbImportUtil::RowMap& rowmap)
{
const NdbImportUtil::RangeList& ranges = rowmap.m_ranges;
const NdbImportUtil::Range* r = ranges.front();
uint i = 0;
while (r != 0)
{
out << i << ": " << *r << endl;
r = r->next();
i++;
}
return out;
}
// pseudo-tables
void
NdbImportUtil::add_pseudo_tables()
{
add_result_table();
add_reject_table();
add_rowmap_table();
add_stopt_table();
add_stats_table();
}
void
NdbImportUtil::add_result_table()
{
Table& table = c_result_table;
table.m_tabid = g_result_tabid;
require(table.m_recsize == 0);
table.add_pseudo_attr("runno",
NdbDictionary::Column::Unsigned);
table.add_pseudo_attr("name",
NdbDictionary::Column::Varchar,
10);
table.add_pseudo_attr("desc",
NdbDictionary::Column::Varchar,
100);
table.add_pseudo_attr("result",
NdbDictionary::Column::Unsigned);
table.add_pseudo_attr("rows",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("reject",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("temperrors",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("runtime",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("utime",
NdbDictionary::Column::Bigunsigned);
add_error_attrs(table);
}
void
NdbImportUtil::add_reject_table()
{
Table& table = c_reject_table;
table.m_tabid = g_reject_tabid;
require(table.m_recsize == 0);
table.add_pseudo_attr("runno",
NdbDictionary::Column::Unsigned);
table.add_pseudo_attr("rowid",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("linenr",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("startpos",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("endpos",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("bytes",
NdbDictionary::Column::Bigunsigned);
add_error_attrs(table);
table.add_pseudo_attr("reject",
NdbDictionary::Column::Text);
}
void
NdbImportUtil::add_rowmap_table()
{
Table& table = c_rowmap_table;
table.m_tabid = g_rowmap_tabid;
require(table.m_recsize == 0);
table.add_pseudo_attr("runno",
NdbDictionary::Column::Unsigned);
table.add_pseudo_attr("start",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("end",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("rows",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("startpos",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("endpos",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("bytes",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("reject",
NdbDictionary::Column::Bigunsigned);
}
void
NdbImportUtil::add_stopt_table()
{
Table& table = c_stopt_table;
table.m_tabid = g_stopt_tabid;
require(table.m_recsize == 0);
table.add_pseudo_attr("runno",
NdbDictionary::Column::Unsigned);
table.add_pseudo_attr("option",
NdbDictionary::Column::Varchar,
100);
table.add_pseudo_attr("value",
NdbDictionary::Column::Unsigned);
}
void
NdbImportUtil::add_stats_table()
{
Table& table = c_stats_table;
table.m_tabid = g_stats_tabid;
require(table.m_recsize == 0);
table.add_pseudo_attr("runno",
NdbDictionary::Column::Unsigned);
table.add_pseudo_attr("id",
NdbDictionary::Column::Unsigned);
table.add_pseudo_attr("name",
NdbDictionary::Column::Varchar,
100);
table.add_pseudo_attr("parent",
NdbDictionary::Column::Unsigned);
table.add_pseudo_attr("obs",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("sum",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("mean",
NdbDictionary::Column::Double);
table.add_pseudo_attr("min",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("max",
NdbDictionary::Column::Bigunsigned);
table.add_pseudo_attr("stddev",
NdbDictionary::Column::Double);
}
#include <math.h>
void
NdbImportUtil::add_error_attrs(Table& table)
{
table.add_pseudo_attr("errortype",
NdbDictionary::Column::Varchar,
10);
table.add_pseudo_attr("errorcode",
NdbDictionary::Column::Unsigned);
table.add_pseudo_attr("sourceline",
NdbDictionary::Column::Unsigned);
table.add_pseudo_attr("errortext",
NdbDictionary::Column::Longvarchar,
1024);
}
void
NdbImportUtil::set_result_row(Row* row,
uint32 runno,
const char* name,
const char* desc,
uint64 rows,
uint64 reject,
uint64 temperrors,
uint64 runtime,
uint64 utime,
const Error& error)
{
const Table& table = c_result_table;
const Attrs& attrs = table.m_attrs;
uint id = 0;
// runno
{
const Attr& attr = attrs[id];
attr.set_value(row, &runno, sizeof(runno));
id++;
}
// name
{
const Attr& attr = attrs[id];
attr.set_value(row, name, strlen(name));
id++;
}
// desc
{
const Attr& attr = attrs[id];
attr.set_value(row, desc, strlen(desc));
id++;
}
// result
{
const Attr& attr = attrs[id];
uint32 value = m_util.has_error(error);
attr.set_value(row, &value, sizeof(value));
id++;
}
// rows
{
const Attr& attr = attrs[id];
attr.set_value(row, &rows, sizeof(rows));
id++;
}
// reject
{
const Attr& attr = attrs[id];
attr.set_value(row, &reject, sizeof(reject));
id++;
}
// temperrors
{
const Attr& attr = attrs[id];
attr.set_value(row, &temperrors, sizeof(temperrors));
id++;
}
// runtime
{
const Attr& attr = attrs[id];
attr.set_value(row, &runtime, sizeof(runtime));
id++;
}
// utime
{
const Attr& attr = attrs[id];
attr.set_value(row, &utime, sizeof(utime));
id++;
}
// error
set_error_attrs(row, table, error, id);
require(id == attrs.size());
}
void
NdbImportUtil::set_reject_row(Row* row,
uint32 runno,
const Error& error,
const char* reject,
uint rejectlen)
{
const Table& table = c_reject_table;
const Attrs& attrs = table.m_attrs;
uint id = 0;
// runno
{
const Attr& attr = attrs[id];
attr.set_value(row, &runno, sizeof(runno));
id++;
}
// rowid
{
const Attr& attr = attrs[id];
attr.set_value(row, &row->m_rowid, sizeof(row->m_rowid));
id++;
}
// linenr
{
const Attr& attr = attrs[id];
attr.set_value(row, &row->m_linenr, sizeof(row->m_linenr));
id++;
}
// startpos
{
const Attr& attr = attrs[id];
attr.set_value(row, &row->m_startpos, sizeof(row->m_startpos));
id++;
}
// endpos
{
const Attr& attr = attrs[id];
attr.set_value(row, &row->m_endpos, sizeof(row->m_endpos));
id++;
}
// bytes
{
const Attr& attr = attrs[id];
uint64 bytes = row->m_endpos - row->m_startpos;
attr.set_value(row, &bytes, sizeof(bytes));
id++;
}
// error
set_error_attrs(row, table, error, id);
// reject
{
const Attr& attr = attrs[id];
attr.set_blob(row, reject, rejectlen);
id++;
}
require(id == attrs.size());
}
void
NdbImportUtil::set_rowmap_row(Row* row,
uint32 runno,
const Range& range)
{
const Table& table = c_rowmap_table;
const Attrs& attrs = table.m_attrs;
uint id = 0;
// runno
{
const Attr& attr = attrs[id];
attr.set_value(row, &runno, sizeof(runno));
id++;
}
// start
{
const Attr& attr = attrs[id];
attr.set_value(row, &range.m_start, sizeof(range.m_start));
id++;
}
// end
{
const Attr& attr = attrs[id];
attr.set_value(row, &range.m_end, sizeof(range.m_end));
id++;
}
// rows
{
const Attr& attr = attrs[id];
uint64 rows = range.m_end - range.m_start;
attr.set_value(row, &rows, sizeof(rows));
id++;
}
// startpos
{
const Attr& attr = attrs[id];
attr.set_value(row, &range.m_startpos, sizeof(range.m_startpos));
id++;
}
// end
{
const Attr& attr = attrs[id];
attr.set_value(row, &range.m_endpos, sizeof(range.m_endpos));
id++;
}
// bytes
{
const Attr& attr = attrs[id];
uint64 bytes = range.m_endpos - range.m_startpos;
attr.set_value(row, &bytes, sizeof(bytes));
id++;
}
// reject
{
const Attr& attr = attrs[id];
attr.set_value(row, &range.m_reject, sizeof(range.m_reject));
id++;
}
require(id == attrs.size());
}
void
NdbImportUtil::set_stopt_row(Row* row,
uint32 runno,
const char* option,
uint32 value)
{
const Table& table = c_stopt_table;
const Attrs& attrs = table.m_attrs;
uint id = 0;
// runno
{
const Attr& attr = attrs[id];
attr.set_value(row, &runno, sizeof(runno));
id++;
}
// option
{
const Attr& attr = attrs[id];
uint len = strlen(option);
attr.set_value(row, option, len);
id++;
}
// value
{
const Attr& attr = attrs[id];
attr.set_value(row, &value, sizeof(value));
id++;
}
require(id == attrs.size());
}
void
NdbImportUtil::set_stats_row(Row* row,
uint32 runno,
const Stat& stat,
bool global)
{
const Table& table = c_stats_table;
const Attrs& attrs = table.m_attrs;
const uint g_offset = !global ? 0 : 1000;
const char* g_prefix = !global ? 0 : "g";
// floats
double obsf = (double)stat.m_obs;
double sum1 = stat.m_sum1;
double sum2 = stat.m_sum2;
uint id = 0;
// runno
{
const Attr& attr = attrs[id];
attr.set_value(row, &runno, sizeof(runno));
id++;
}
// id
{
const Attr& attr = attrs[id];
uint idval = stat.m_id + g_offset;
attr.set_value(row, &idval, sizeof(idval));
id++;
}
// name
{
const Attr& attr = attrs[id];
if (g_prefix == 0)
{
uint namelen = strlen(stat.m_name);
attr.set_value(row, stat.m_name, namelen);
}
else
{
Name name(g_prefix, stat.m_name);
uint namelen = strlen(name.str());
attr.set_value(row, name.str(), namelen);
}
id++;
}
// parent
{
const Attr& attr = attrs[id];
uint parentval = stat.m_id == 0 ? stat.m_parent : stat.m_parent + g_offset;
attr.set_value(row, &parentval, sizeof(parentval));
id++;
}
// obs
{
const Attr& attr = attrs[id];
attr.set_value(row, &stat.m_obs, sizeof(stat.m_obs));
id++;
}
// sum
{
const Attr& attr = attrs[id];
attr.set_value(row, &stat.m_sum, sizeof(stat.m_sum));
id++;
}
// mean
{
const Attr& attr = attrs[id];
double mean = 0.0;
if (stat.m_obs != 0)
mean = sum1 / obsf;
if (!std::isfinite(mean))
mean = 0.0;
attr.set_value(row, &mean, sizeof(mean));
id++;
}
// min
{
const Attr& attr = attrs[id];
attr.set_value(row, &stat.m_min, sizeof(stat.m_min));
id++;
}
// max
{
const Attr& attr = attrs[id];
attr.set_value(row, &stat.m_max, sizeof(stat.m_max));
id++;
}
// stddev
{
const Attr& attr = attrs[id];
double stddev = 0.0;
if (stat.m_obs != 0)
stddev = ::sqrt((obsf * sum2 - (sum1 * sum1)) / (obsf * obsf));
if (!std::isfinite(stddev))
stddev = 0.0;
attr.set_value(row, &stddev, sizeof(stddev));
id++;
}
require(id == attrs.size());
}
void
NdbImportUtil::set_error_attrs(Row* row,
const Table& table,
const Error& error,
uint& id)
{
const Attrs& attrs = table.m_attrs;
// errortype
{
const Attr& attr = attrs[id];
const char* errortype = error.gettypetext();
attr.set_value(row, errortype, strlen(errortype));
id++;
}
// errorcode
{
const Attr& attr = attrs[id];
int32 errorcode = error.code;
attr.set_value(row, &errorcode, sizeof(errorcode));
id++;
}
// sourceline
{
const Attr& attr = attrs[id];
uint32 errorline = error.line;
attr.set_value(row, &errorline, sizeof(errorline));
id++;
}
// errortext
{
const Attr& attr = attrs[id];
attr.set_value(row, error.text, strlen(error.text));
id++;
}
}
// buf
NdbImportUtil::Buf::Buf(bool split) :
m_split(split)
{
m_allocptr = 0;
m_allocsize = 0;
m_data = 0;
m_size = 0;
m_top = 0;
m_start = 0;
m_tail = 0;
m_len = 0;
m_eof = false;
m_pos = 0;
m_lineno = 0;
}
NdbImportUtil::Buf::~Buf()
{
delete [] m_allocptr;
}
void
NdbImportUtil::Buf::alloc(uint pagesize, uint pagecnt)
{
require(m_allocptr == 0);
require(pagesize != 0 && (pagesize & (pagesize - 1)) == 0);
require(pagecnt != 0);
uint size = pagesize * pagecnt;
uint allocsize = size + (pagesize - 1) + 1;
uchar* allocptr = new uchar [allocsize];
uchar* data = allocptr;
uint misalign = (UintPtr)data & (pagesize - 1);
if (misalign != 0)
{
data += (pagesize - misalign);
uint misalign2 = (UintPtr)data & (pagesize - 1);
require(misalign2 == 0);
}
require(data + size < allocptr + allocsize);
m_allocptr = allocptr;
m_allocsize = allocsize;
m_data = data;
m_size = size;
m_top = 0;
m_start = 0;
m_tail = 0;
m_len = 0;
if (m_split)
{
require(pagecnt % 2 == 0);
m_top = size / 2;
m_start = m_top;
}
}
void
NdbImportUtil::Buf::copy(const uchar* src, uint len)
{
require(m_start + m_len + len < m_allocsize);
memcpy(&m_data[m_start + m_len], src, len);
m_len += len;
m_data[m_start + m_len] = 0;
}
void
NdbImportUtil::Buf::reset()
{
m_start = 0;
if (m_split)
{
require(2 * m_top == m_size);
m_start = m_top;
}
m_tail = 0;
m_len = 0;
m_eof = false;
m_pos = 0;
m_lineno = 0;
}
int
NdbImportUtil::Buf::movetail(Buf& dst)
{
require(m_tail <= m_len);
uint bytes = m_len - m_tail;
if (bytes > dst.m_start)
{
return -1;
}
const uchar* srcptr = &m_data[m_start + m_tail];
uchar* dstptr = &dst.m_data[dst.m_start - bytes];
memcpy(dstptr, srcptr, bytes);
m_len = m_tail;
dst.m_start -= bytes;
dst.m_len += bytes;
return 0;
}
NdbOut&
operator<<(NdbOut& out, const NdbImportUtil::Buf& buf)
{
out << "allocsize=" << buf.m_allocsize;
out << " size=" << buf.m_size;
out << " top=" << buf.m_top;
out << " start=" << buf.m_start;
out << " tail=" << buf.m_tail;
out << " len=" << buf.m_len;
out << " eof=" << buf.m_eof;
const uchar* dataptr = &buf.m_data[buf.m_start];
{
char dst[100];
uint len = buf.m_len;
if (len > 10)
len = 10;
NdbImportUtil::pretty_print(dst, &dataptr[0], len);
out << " buf=" << "0:" << dst;
}
{
char dst[100];
require(buf.m_len >= buf.m_pos);
uint n = buf.m_len - buf.m_pos;
if (n > 10)
n = 10;
NdbImportUtil::pretty_print(dst, &dataptr[buf.m_pos], n);
out << " pos=" << buf.m_pos << ":" << dst;
}
{
out << " lineno=" << buf.m_lineno;
}
return out;
}
void
NdbImportUtil::pretty_print(char* dst, const void* ptr, uint len)
{
const char* p = (const char*)ptr;
char *s = dst;
for (uint i = 0; i < len; i++)
{
int c = p[i];
if (isascii(c) && isprint(c))
sprintf(s, "%c", c);
else if (c == '\n')
sprintf(s, "%s", "\\n");
else
sprintf(s, "<%02X>", (uchar)c);
s += strlen(s);
}
}
// files
NdbImportUtil::File::File(NdbImportUtil& util, Error& error) :
m_util(util),
m_error(error)
{
m_fd = -1;
m_flags = 0;
}
NdbImportUtil::File::~File()
{
if (m_fd != -1)
(void)::close(m_fd);
}
int
NdbImportUtil::File::do_open(int flags)
{
const char* path = get_path();
require(m_fd == -1);
#ifndef _WIN32
int fd = ::open(path, flags, Creat_mode);
#else
int fd = ::_open(path, flags, Creat_mode);
#endif
if (fd == -1)
{
const char* type = "unknown";
if (flags == Read_flags)
type = "read";
if (flags == Write_flags)
type = "write";
if (flags == Append_flags)
type = "append";
m_util.set_error_os(m_error, __LINE__,
"%s: open for %s failed", path, type);
return -1;
}
m_fd = fd;
m_flags = flags;
return 0;
}
int
NdbImportUtil::File::do_read(uchar* dst, uint size, uint& len)
{
const char* path = get_path();
require(m_fd != -1);
len = 0;
while (len < size)
{
// short read is possible on pipe
#ifndef _WIN32
int ret = ::read(m_fd, dst + len, size - len);
#else
int ret = ::_read(m_fd, dst + len, size - len);
#endif
if (ret == -1)
{
m_util.set_error_os(m_error, __LINE__,
"%s: read %u bytes failed", path, size - len);
return -1;
}
uint n = (uint)ret;
if (n == 0)
break;
len += n;
require(len <= size);
}
return 0;
}
int
NdbImportUtil::File::do_read(Buf& buf)
{
uint dstpos = buf.m_start + buf.m_len;
require(dstpos == buf.m_top);
require(dstpos <= buf.m_size);
uchar* dst = buf.m_data + dstpos;
uint size = buf.m_size - dstpos;
uint len = 0;
if (do_read(dst, size, len) == -1)
return -1;
buf.m_eof = (len == 0);
buf.m_len += len;
uint endpos = buf.m_start + buf.m_len;
require(endpos <= buf.m_size);
require(endpos < buf.m_allocsize);
buf.m_data[endpos] = 0;
return 0;
}
int
NdbImportUtil::File::do_write(const uchar* src, uint size)
{
const char* path = get_path();
require(m_fd != -1);
#ifndef _WIN32
int ret = ::write(m_fd, src, size);
#else
int ret = ::_write(m_fd, src, size);
#endif
if (ret == -1)
{
m_util.set_error_os(m_error, __LINE__,
"%s: write %u bytes failed", path, size);
return -1;
}
// short write is considered error
if (uint(ret) != size)
{
m_util.set_error_os(m_error, __LINE__,
"%s: short write %u < %u", path, (uint)ret, size);
return -1;
}
return 0;
}
int
NdbImportUtil::File::do_write(const Buf& buf)
{
const uchar* src = &buf.m_data[buf.m_start];
uint len = buf.m_len;
if (do_write(src, len) == -1)
return -1;
return 0;
}
int
NdbImportUtil::File::do_close()
{
const char* path = get_path();
if (m_fd == -1)
return 0;
#ifndef _WIN32
if (::close(m_fd) == -1)
#else
if (::_close(m_fd) == -1)
#endif
{
m_util.set_error_os(m_error, __LINE__,
"%s: close failed", path);
return -1;
}
m_fd = -1;
return 0;
}
int
NdbImportUtil::File::do_seek(uint64 offset)
{
const char* path = get_path();
require(m_fd != -1);
#ifndef _WIN32
off_t off = (off_t)offset;
if (::lseek(m_fd, off, SEEK_SET) == -1)
#else
__int64 off = (__int64)offset;
if (::_lseeki64(m_fd, off, SEEK_SET) == -1)
#endif
{
m_util.set_error_os(m_error, __LINE__,
"%s: lseek %" PRIu64 " failed", path, offset);
return -1;
}
return 0;
}
// stats
NdbImportUtil::Stat::Stat(Stats& stats,
uint id,
const char* name,
uint parent,
uint level,
uint flags) :
m_stats(stats),
m_id(id),
m_name(name),
m_parent(parent),
m_level(level),
m_flags(flags)
{
m_childcnt = 0;
m_firstchild = StatNULL;
m_lastchild = StatNULL;
m_nextchild = StatNULL;
if (m_parent != StatNULL)
{
Stat* parentstat = m_stats.get(m_parent);
if (parentstat->m_childcnt == 0)
{
parentstat->m_firstchild = m_id;
parentstat->m_lastchild = m_id;
}
else
{
Stat* lastchildstat = m_stats.get(parentstat->m_lastchild);
require(lastchildstat->m_nextchild == StatNULL);
lastchildstat->m_nextchild = m_id;
parentstat->m_lastchild = m_id;
}
parentstat->m_childcnt++;
}
reset();
}
void
NdbImportUtil::Stat::add(uint64 val)
{
uint id = m_id;
do
{
Stat* stat = m_stats.get(id);
stat->m_obs++;
stat->m_sum += val;
if (stat->m_obs == 1)
{
stat->m_min = val;
stat->m_max = val;
}
else
{
if (stat->m_min > val)
stat->m_min = val;
if (stat->m_max < val)
stat->m_max = val;
}
stat->m_sum1 += double(val);
stat->m_sum2 += double(val) * double(val);
id = stat->m_parent;
// root level stats are not useful
} while (id != 0);
}
void
NdbImportUtil::Stat::reset()
{
m_obs = 0;
m_sum = 0;
m_min = 0;
m_max = 0;
m_sum1 = 0.0;
m_sum2 = 0.0;
}
NdbImportUtil::Stats::Stats(NdbImportUtil& util) :
m_util(util)
{
Stat* rootstat = new Stat(*this, 0, "root", StatNULL, 0, 0);
m_stats.push_back(rootstat);
validate();
}
NdbImportUtil::Stats::~Stats()
{
for (uint i = 0; i < m_stats.size(); i++)
{
Stat* stat = get(i);
delete stat;
}
}
NdbImportUtil::Stat*
NdbImportUtil::Stats::create(const char* name, uint parent, uint flags)
{
lock();
{
Stat* stat = find(name);
if (stat != 0)
{
log_debug(2, "use existing " << stat->m_name << " id=" << stat->m_id);
unlock();
return stat;
}
}
Stat* parentstat = get(parent);
uint parentlevel = parentstat->m_level;
uint id = m_stats.size();
Stat* stat = new Stat(*this, id, name, parent, parentlevel + 1, flags);
m_stats.push_back(stat);
log_debug(2, "created stat id=" << stat->m_id << " name=" << stat->m_name);
validate();
unlock();
return stat;
}
NdbImportUtil::Stat*
NdbImportUtil::Stats::get(uint i)
{
require(i < m_stats.size());
Stat* stat = m_stats[i];
require(stat != 0);
require(stat->m_id == i);
return stat;
}
const NdbImportUtil::Stat*
NdbImportUtil::Stats::get(uint i) const
{
require(i < m_stats.size());
const Stat* stat = m_stats[i];
require(stat != 0);
require(stat->m_id == i);
return stat;
}
NdbImportUtil::Stat*
NdbImportUtil::Stats::find(const char* name) const
{
for (uint i = 0; i < m_stats.size(); i++)
{
Stat* stat = m_stats[i];
require(stat != 0);
if (strcmp(stat->m_name, name) == 0)
return stat;
}
return 0;
}
void
NdbImportUtil::Stats::add(uint id, uint64 val)
{
Stat* stat = get(id);
stat->add(val);
}
const NdbImportUtil::Stat*
NdbImportUtil::Stats::next(uint id) const
{
require(id < m_stats.size());
const Stat* stat = m_stats[id];
require(stat != 0);
if (stat->m_firstchild != StatNULL)
{
stat = get(stat->m_firstchild);
return stat;
}
while (1)
{
if (stat->m_nextchild != StatNULL)
{
stat = get(stat->m_nextchild);
return stat;
}
if (stat->m_parent == StatNULL)
{
break;
}
stat = get(stat->m_parent);
}
return 0;
}
void
NdbImportUtil::Stats::reset()
{
for (uint i = 0; i < m_stats.size(); i++)
{
Stat* stat = get(i);
stat->reset();
}
}
#if defined(VM_TRACE) || defined(TEST_NDBIMPORTUTIL)
void
NdbImportUtil::Stats::validate() const
{
bool* seen = new bool [m_stats.size()];
Validate v(StatNULL, 0, 0, seen);
for (uint i = 0; i < m_stats.size(); i++)
v.m_seen[i] = false;
(void)validate(v);
for (uint i = 0; i < m_stats.size(); i++)
require(v.m_seen[i] == true);
delete [] seen;
}
const NdbImportUtil::Stat*
NdbImportUtil::Stats::validate(Validate& v) const
{
const Stat* stat = get(v.m_id);
require(stat->m_parent == v.m_parent);
require(stat->m_id == v.m_id);
require(stat->m_level == v.m_level);
const Stat* stat2 = find(stat->m_name);
require(stat == stat2);
require(v.m_seen[v.m_id] == false);
v.m_seen[v.m_id] = true;
uint sibling = stat->m_nextchild;
if (sibling != StatNULL)
{
Validate v2(v.m_parent, sibling, v.m_level, v.m_seen);
validate(v2);
}
uint child = stat->m_firstchild;
if (child != StatNULL)
{
Validate v2(v.m_id, child, v.m_level + 1, v.m_seen);
validate(v2);
}
return stat;
}
#endif
NdbOut&
operator<<(NdbOut& out, const NdbImportUtil::Stats& stats)
{
out << "stats ";
return out;
}
// timer
NdbImportUtil::Timer::Timer()
{
// make sure initialized to avoid assert
start();
stop();
}
void
NdbImportUtil::Timer::start()
{
m_start = NdbTick_getCurrentTicks();
}
void
NdbImportUtil::Timer::stop()
{
m_stop = NdbTick_getCurrentTicks();
if (NdbTick_Compare(m_start, m_stop) > 0)
{
// not worth crashing
m_start = m_stop;
}
struct ndb_rusage ru;
if (Ndb_GetRUsage(&ru, false) == 0)
{
m_utime_msec = ru.ru_utime / 1000;
m_stime_msec = ru.ru_stime / 1000;
}
}
uint64
NdbImportUtil::Timer::elapsed_sec() const
{
return NdbTick_Elapsed(m_start, m_stop).seconds();
}
uint64
NdbImportUtil::Timer::elapsed_msec() const
{
return NdbTick_Elapsed(m_start, m_stop).milliSec();
}
uint64
NdbImportUtil::Timer::elapsed_usec() const
{
return NdbTick_Elapsed(m_start, m_stop).microSec();
}
// error
void
NdbImportUtil::set_error_gen(Error& error, int line,
const char* fmt, ...)
{
c_error_lock.lock();
new (&error) Error;
error.line = line;
error.type = Error::Type_gen;
if (fmt != 0)
{
va_list ap;
va_start(ap, fmt);
vsnprintf(error.text, sizeof(error.text), fmt, ap);
va_end(ap);
}
log_debug(1, "E " << error);
if (c_opt.m_abort_on_error)
abort();
c_error_lock.unlock();
}
void
NdbImportUtil::set_error_usage(Error& error, int line,
const char* fmt, ...)
{
c_error_lock.lock();
new (&error) Error;
error.line = line;
error.type = Error::Type_usage;
if (fmt != 0)
{
va_list ap;
va_start(ap, fmt);
vsnprintf(error.text, sizeof(error.text), fmt, ap);
va_end(ap);
}
log_debug(1, "E " << error);
if (c_opt.m_abort_on_error)
abort();
c_error_lock.unlock();
}
void
NdbImportUtil::set_error_alloc(Error& error, int line)
{
c_error_lock.lock();
new (&error) Error;
error.line = line;
error.type = Error::Type_alloc;
log_debug(1, "E " << error);
if (c_opt.m_abort_on_error)
abort();
c_error_lock.unlock();
}
void
NdbImportUtil::set_error_mgm(Error& error, int line,
NdbMgmHandle handle)
{
c_error_lock.lock();
new (&error) Error;
error.line = line;
error.type = Error::Type_mgm;
error.code = ndb_mgm_get_latest_error(handle);
snprintf(error.text, sizeof(error.text),
"%s", ndb_mgm_get_latest_error_msg(handle));
log_debug(1, "E " << error);
if (c_opt.m_abort_on_error)
abort();
c_error_lock.unlock();
}
void
NdbImportUtil::set_error_con(Error& error, int line,
const Ndb_cluster_connection* con)
{
c_error_lock.lock();
new (&error) Error;
error.line = line;
error.type = Error::Type_con;
error.code = con->get_latest_error();
snprintf(error.text, sizeof(error.text), "%s", con->get_latest_error_msg());
log_debug(1, "E " << error);
if (c_opt.m_abort_on_error)
abort();
c_error_lock.unlock();
}
void
NdbImportUtil::set_error_ndb(Error& error, int line,
const NdbError& ndberror, const char* fmt, ...)
{
c_error_lock.lock();
new (&error) Error;
error.line = line;
error.type = Error::Type_ndb;
error.code = ndberror.code;
snprintf(error.text, sizeof(error.text), "%s", ndberror.message);
log_debug(1, "E " << error);
if (c_opt.m_abort_on_error)
abort();
c_error_lock.unlock();
}
void
NdbImportUtil::set_error_os(Error& error, int line,
const char* fmt, ...)
{
c_error_lock.lock();
new (&error) Error;
error.line = line;
error.type = Error::Type_os;
error.code = errno;
if (fmt != 0)
{
va_list ap;
va_start(ap, fmt);
vsnprintf(error.text, sizeof(error.text), fmt, ap);
va_end(ap);
}
{
uint len = strlen(error.text);
if (len < sizeof(error.text))
{
uint left = sizeof(error.text) - len;
snprintf(&error.text[len], left, ": errno=%u: %s",
errno, strerror(errno));
}
}
log_debug(1, "E " << error);
if (c_opt.m_abort_on_error)
abort();
c_error_lock.unlock();
}
void
NdbImportUtil::set_error_data(Error& error, int line,
int code, const char* fmt, ...)
{
c_error_lock.lock();
new (&error) Error;
error.line = line;
error.type = Error::Type_data;
error.code = code;
if (fmt != 0)
{
va_list ap;
va_start(ap, fmt);
vsnprintf(error.text, sizeof(error.text), fmt, ap);
va_end(ap);
}
log_debug(1, "E " << error);
if (c_opt.m_abort_on_error)
abort();
c_error_lock.unlock();
}
void
NdbImportUtil::copy_error(Error& error, const Error& error2)
{
error.type = error2.type;
error.code = error2.code;
error.line = error2.line;
memcpy(error.text, error2.text, sizeof(error.text));
}
bool NdbImportUtil::g_stop_all = false;
void
NdbImportUtil::fmt_msec_to_hhmmss(char* str, uint64 msec)
{
uint sec = msec / 1000;
uint hh = sec / 3600;
sec -= hh * 3600;
uint mm = sec / 60;
sec -= mm * 60;
uint ss = sec;
uint ff = msec - sec * 1000;
(void)ff;
sprintf(str, "%uh%um%us", hh, mm, ss);
}
// unittest
#ifdef TEST_NDBIMPORTUTIL
typedef NdbImportUtil::ListEnt UtilListEnt;
typedef NdbImportUtil::List UtilList;
typedef NdbImportUtil::Table UtilTable;
typedef NdbImportUtil::RowCtl UtilRowCtl;
typedef NdbImportUtil::Row UtilRow;
typedef NdbImportUtil::RowList UtilRowList;
typedef NdbImportUtil::Range UtilRange;
typedef NdbImportUtil::RangeList UtilRangeList;
typedef NdbImportUtil::RowMap UtilRowMap;
typedef NdbImportUtil::Buf UtilBuf;
typedef NdbImportUtil::File UtilFile;
typedef NdbImportUtil::Name UtilName;
typedef NdbImportUtil::Stat UtilStat;
typedef NdbImportUtil::Stats UtilStats;
#include <NdbTap.hpp>
#include <ndb_rand.h>
#include <NdbEnv.h>
// increase size of some tests if release-compiled
static bool mybigtest = false;
static uint
myrandom()
{
return (uint)ndb_rand() * (uint)ndb_rand(); // < 2**30
}
static uint
myrandom(uint m)
{
require(m != 0);
uint n = myrandom();
return n % m;
}
struct MyRec : UtilListEnt {
uint m_index;
bool m_member;
MyRec() {
m_index = Inval_uint;
m_member = false;
}
};
struct MyRecs : UtilList {
};
static int
testlist()
{
ndbout << "testlist" << endl;
NdbImportUtil util;
util.c_opt.m_log_level = 3;
MyRecs recs;
const uint poolsize = 256;
MyRec* recpool = new MyRec [poolsize];
for (uint n = 0; n < poolsize; n++)
{
MyRec* rec = &recpool[n];
rec->m_index = n;
}
const uint numops = 1024 * poolsize;
uint ops[5];
ops[0] = 0; // push back
ops[1] = 0; // pop front
ops[2] = 0; // remove
ops[3] = 0; // push after
ops[4] = 0; // push before
uint max_occup = 0;
for (uint numop = 0; numop < numops; numop++)
{
MyRec* rec = 0;
while (1)
{
uint n = myrandom(poolsize);
rec = &recpool[n];
if (rec->m_member)
{
if (myrandom(100) < 80)
continue;
}
break;
}
if (!rec->m_member)
{
recs.push_back(rec);
rec->m_member = true;
ops[0]++;
}
else if (myrandom(100) < 50)
{
MyRec* rec = static_cast<MyRec*>(recs.pop_front());
require(rec != 0);
rec->m_member = false;
ops[1]++;
}
else if (myrandom(100) < 50)
{
recs.remove(rec);
rec->m_member = false;
ops[2]++;
}
else
{
uint n2 = myrandom(poolsize);
MyRec* rec2 = &recpool[n2];
if (!rec2->m_member)
{
if (myrandom(100) < 50)
{
recs.push_after(rec, rec2);
rec2->m_member = true;
ops[3]++;
}
else
{
recs.push_before(rec, rec2);
rec2->m_member = true;
ops[4]++;
}
}
}
if (max_occup < recs.m_cnt)
max_occup = recs.m_cnt;
}
uint last_occup = recs.m_cnt;
for (MyRec* rec = static_cast<MyRec*>(recs.m_front);
rec != 0;
rec = static_cast<MyRec*>(rec->m_next))
{
uint n = rec->m_index;
require(n < poolsize);
require(rec == &recpool[n]);
require(rec->m_member);
}
for (uint n = 0; n < poolsize; n++)
{
MyRec* rec = &recpool[n];
if (rec->m_member)
{
recs.remove(rec);
rec->m_member = false;
}
}
require(recs.m_cnt == 0);
uint ins = ops[0] + ops[3] + ops[4];
uint del = ops[1] + ops[2];
require(last_occup == ins - del);
delete [] recpool;
ndbout << "max_occup=" << max_occup << endl;
ndbout << "last_occup=" << last_occup << endl;
ndbout << "push_back: " << ops[0] << endl;
ndbout << "pop_front: " << ops[1] << endl;
ndbout << "remove: " << ops[2] << endl;
ndbout << "push_after: " << ops[3] << endl;
ndbout << "push_before: " << ops[4] << endl;
return 0;
}
static int
testrowlist1()
{
ndbout << "testrowlist1" << endl;
NdbImportUtil util;
UtilTable table;
table.add_pseudo_attr("a", NdbDictionary::Column::Unsigned);
table.add_pseudo_attr("b", NdbDictionary::Column::Varchar, 10);
const uint loops = !mybigtest ? 100 : 1000;
const uint rows = !mybigtest ? 1000 : 10000;
for (uint loop = 0; loop < loops; loop++)
{
UtilRowList list1;
UtilRowList list2;
while (list1.cnt() < rows)
{
uint cnt = 1 + myrandom(rows - list1.cnt());
util.alloc_rows(table, cnt, list1);
}
require(list1.cnt() == rows);
list2.push_back_from(list1);
require(list1.cnt() == 0);
require(list2.cnt() == rows);
util.free_rows(list2);
require(list2.cnt() == 0);
require(util.c_rows_free->cnt() == rows);
}
return 0;
}
static int
testrowlist2()
{
ndbout << "testrowlist2" << endl;
NdbImportUtil util;
UtilTable table;
table.add_pseudo_attr("a", NdbDictionary::Column::Unsigned);
table.add_pseudo_attr("b", NdbDictionary::Column::Varchar, 10);
const uint loops = !mybigtest ? 100 : 1000;
const uint rows = !mybigtest ? 1000 : 10000;
{
UtilRowList list;
util.alloc_rows(table, rows, list);
util.free_rows(list);
}
for (uint loop = 0; loop < loops; loop++)
{
UtilRowList list1;
UtilRowList list2;
list1.m_rowbatch = 1 + myrandom(rows);
list2.m_rowbatch = 1 + myrandom(rows);
const uint cnt = myrandom(rows + 1);
util.alloc_rows(table, cnt, list1);
require(list1.cnt() == cnt);
do
{
uint timeout = myrandom(10) != 0 ? 0 : 10;
UtilRowCtl ctl(timeout);
list2.push_back_from(list1, ctl);
util.free_rows(list2);
} while (list1.cnt() != 0);
require(util.c_rows_free->cnt() == rows);
util.alloc_rows(table, cnt, list1);
require(list1.cnt() == cnt);
do
{
uint timeout = myrandom(10) != 0 ? 0 : 10;
UtilRowCtl ctl(timeout);
list1.pop_front_to(list2, ctl);
util.free_rows(list2);
} while (list1.cnt() != 0);
require(util.c_rows_free->cnt() == rows);
}
return 0;
}
static int
testrowmap1()
{
ndbout << "testrowmap1" << endl;
NdbImportUtil util;
/*
* Create random ascending ranges. This does not represent
* a valid rowmap because gaps between ranges can be zero.
*/
const uint maxranges = 1000;
const uint numranges = myrandom(maxranges);
ndbout << "numranges = " << numranges << endl;
const uint maxgap = 5;
const uint maxcount = 10;
const uint maxrowid = maxranges * (maxcount + maxgap);
UtilRange tstranges[maxranges];
bool rowexist[maxrowid];
uint toprowid = 0;
for (uint k = 0; k < maxrowid; k++)
rowexist[k] = false;
{
uint start = 0;
for (uint i = 0; i < numranges; i++)
{
UtilRange& r = tstranges[i];
uint gap = myrandom(maxgap + 1);
uint count = 1 + myrandom(maxcount);
r.m_start = start + gap;
r.m_end = r.m_start + count;
r.m_reject = myrandom(1 + count);
start = r.m_end;
for (uint k = r.m_start; k < r.m_end; k++)
rowexist[k] = true;
toprowid = r.m_end;
}
}
ndbout << "toprowid=" << toprowid << endl;
//
ndbout << "map1: create in ascending order" << endl;
UtilRowMap map1(util);
for (uint i = 0; i < numranges; i++)
{
UtilRange r = tstranges[i];
map1.add(r);
}
ndbout << "map1: " << map1.size() << endl;
for (uint k = 0; k < toprowid; k++)
{
if (rowexist[k])
require(map1.find(k) != 0);
else
require(map1.find(k) == 0);
}
uint reorder[maxranges];
{
for (uint i = 0; i < numranges; i++)
reorder[i] = i;
for (uint i = 0; i < numranges; i++)
{
uint j = myrandom(numranges);
uint k = reorder[i];
reorder[i] = reorder[j];
reorder[j] = k;
}
}
//
ndbout << "map2: create in random order" << endl;
UtilRowMap map2(util);
for (uint i = 0; i < numranges; i++)
{
uint j = reorder[i];
UtilRange r = tstranges[j];
map2.add(r);
}
ndbout << "map2: " << map2.size() << endl;
require(map1.equal(map2));
//
ndbout << "map3: create from 2 random pieces" << endl;
UtilRowMap map3(util);
UtilRowMap map3a(util);
UtilRowMap map3b(util);
for (uint i = 0; i < numranges; i++)
{
UtilRange r = tstranges[i];
if (myrandom(100) < 50)
map3a.add(r);
else
map3b.add(r);
}
ndbout << "map3a: " << map3a.size() << endl;
//ndbout << map3a;
ndbout << "map3b: " << map3b.size() << endl;
//ndbout << map3b;
ndbout << "add map3a" << endl;
map3.add(map3a);
ndbout << "add map3b" << endl;
map3.add(map3b);
ndbout << "map3: " << map3.size() << endl;
//ndbout << map3;
require(map1.equal(map3));
//
ndbout << "map4: delete all in random order" << endl;
UtilRowMap map4(util);
map4.add(map1);
ndbout << "map4: " << map4.size() << endl;
bool rowexist4[maxrowid];
for (uint k = 0; k < maxrowid; k++)
rowexist4[k] = rowexist[k];
while (map4.size() != 0)
{
uint k = myrandom(toprowid);
if (rowexist4[k])
{
require(map4.remove(k) == true);
rowexist4[k] = false;
}
}
ndbout << "map4: " << map4.size() << endl;
return 0;
}
// an old "intrusive" test, fix later
static int
testrowmap2()
{
#if 0
ndbout << "testrowmap2" << endl;
const uint maxranges = 10000;
//
ndbout << "map1: create manually" << endl;
UtilRowMap map1;
UtilRowMap::Ranges& ranges1 = map1.m_ranges;
{
uint64 start = 0;
for (uint i = 0; i < maxranges; i++)
{
UtilRowMap::Range r;
uint64 gap = myrandom(10);
uint64 count = 1 + myrandom(10);
r.m_start = start + gap;
r.m_end = r.m_start + count;
r.m_reject = myrandom(1 + count / 10);
ranges1.push_back(r);
start = r.m_end;
}
}
ndbout << "map1: size " << ranges1.size() << endl;
require(ranges1.size() == maxranges);
//
ndbout << "map2: create via merge from map1" << endl;
UtilRowMap map2;
UtilRowMap::Ranges& ranges2 = map2.m_ranges;
{
uint i = 0;
while (i < maxranges)
{
UtilRowMap::Range r = ranges1[i];
uint j = i + 1;
while (j < maxranges)
{
UtilRowMap::Range r2 = ranges1[j];
if (r.m_end < r2.m_start)
break;
r.m_end += r2.m_end - r2.m_start;
r.m_reject += r2.m_reject;
j++;
}
ranges2.push_back(r);
i = j;
}
}
ndbout << "map2: size " << ranges2.size() << endl;
//
ndbout << "map3: create via util from map1" << endl;
UtilRowMap map3;
UtilRowMap::Ranges& ranges3 = map3.m_ranges;
{
uint* reorder = new uint [maxranges];
for (uint i = 0; i < maxranges; i++)
{
reorder[i] = i;
}
for (uint i = 0; i < maxranges; i++)
{
if (myrandom(10) == 0)
{
uint m = 100;
if (m > maxranges - i)
m = maxranges - i;
uint j = i + myrandom(m);
// std::swap
uint k = reorder[i];
reorder[i] = reorder[j];
reorder[j] = k;
}
}
for (uint i = 0; i < maxranges; i++)
{
uint j = reorder[i];
UtilRowMap::Range r = ranges1[j];
map3.add(r);
}
delete [] reorder;
}
ndbout << "map3: size " << ranges3.size() << endl;
//
ndbout << "map2 vs map3: verify" << endl;
require(ranges2.size() == ranges3.size());
for (uint i = 0; i < ranges2.size(); i++)
{
const UtilRowMap::Range r2 = ranges2[i];
const UtilRowMap::Range r3 = ranges3[i];
require(r2.m_start == r3.m_start);
require(r2.m_end == r3.m_end);
require(r2.m_reject == r2.m_reject);
}
//
ndbout << "mark existing" << endl;
struct Mark {
Mark(uint64 size) {
m_mark = new bool [size];
m_size = size;
for (uint64 i = 0; i < m_size; i++)
m_mark[i] = false;
m_cnt = 0;
}
~Mark() {
delete [] m_mark;
};
void set(uint64 pos, uint64 cnt) {
require(pos + cnt <= m_size);
for(uint64 i = pos; i < pos + cnt; i++) {
require(m_mark[i] == false);
m_mark[i] = true;
m_cnt++;
}
}
bool get(uint64 pos) {
require(pos < m_size);
return m_mark[pos];
}
void del(uint64 pos) {
require(pos < m_size);
require(m_mark[pos] == true);
m_mark[pos] = false;
require(m_cnt != 0);
m_cnt--;
}
bool* m_mark;
uint64 m_size;
uint64 m_cnt;
};
uint64 maxid = ranges3.back().m_end;
Mark mark(maxid);
for (uint i = 0; i < ranges3.size(); i++)
{
const UtilRowMap::Range r3 = ranges3[i];
mark.set(r3.m_start, r3.m_end - r3.m_start);
}
ndbout << "ids: size=" << mark.m_size << " cnt=" << mark.m_cnt << endl;
//
ndbout << "test find" << endl;
for (uint64 id = 0; id < maxid; id++)
{
UtilRowMap::Iterator it;
if (mark.get(id) == false)
require(map3.find(id, it) == false);
if (mark.get(id) == true)
require(map3.find(id, it) == true);
}
//
ndbout << "test remove: random 50%" << endl;
uint64 oldcnt = mark.m_cnt;
while (oldcnt < mark.m_cnt * 2)
{
uint64 id = myrandom() % mark.m_size;
if (mark.get(id))
{
map3.remove(id);
mark.del(id);
}
}
//
ndbout << "test remove: rest " << mark.m_cnt << endl;
for (uint64 id = 0; id < mark.m_size; id++)
{
if (mark.get(id))
{
map3.remove(id);
mark.del(id);
}
}
require(map3.empty());
require(mark.m_cnt == 0);
#endif
return 0;
}
static int
testbuf()
{
ndbout << "testbuf" << endl;
uint loops = 128 * 1024;
uint min_off = UINT_MAX;
uint max_off = 0;
for (uint n = 0; n < loops; n++)
{
UtilBuf* buf = new UtilBuf;
uint pagesize_log2 = 1 + myrandom(15);
uint pagesize = (1 << pagesize_log2);
uint pagecnt = 1 + myrandom(1024);
buf->alloc(pagesize, pagecnt);
uint off = buf->m_data - buf->m_allocptr;
if (min_off > off)
min_off = off;
if (max_off < off)
max_off = off;
delete buf;
}
ndbout << "min_off=" << min_off << " max_off=" << max_off << endl;
return 0;
}
static int
testprint()
{
ndbout << "testprint" << endl;
uchar buf[256];
char dst[5 * 256];
for (uint i = 0; i < 256; i++)
buf[i] = i;
NdbImportUtil::pretty_print(dst, buf, 256);
ndbout << dst << endl;
return 0;
}
static int
testfile()
{
ndbout << "testfile" << endl;
NdbImportUtil util;
const char* path = "test.csv";
struct stat st;
if (stat(path, &st) == -1)
{
ndbout << path << ": skip on errno " << errno << endl;
return 0;
}
for (int split = 0; split <= 1; split++)
{
ndbout << "read " << path << " buf split=" << split << endl;
UtilBuf buf(split);
buf.alloc(4096, 8);
UtilFile file(util, util.c_error);
file.set_path(path);
require(file.do_open(UtilFile::Read_flags) == 0);
uint totlen = 0;
uint totread = 0;
while (1)
{
buf.reset();
int ret = file.do_read(buf);
require(ret == 0);
if (buf.m_eof)
break;
totlen += buf.m_len;
totread++;
}
require(totlen == st.st_size);
ndbout << "len=" << totlen << " reads=" << totread << endl;
require(file.do_close() == 0);
}
return 0;
}
static int
teststat()
{
ndbout << "teststat" << endl;
NdbImportUtil util;
util.c_opt.m_log_level = 3;
UtilStats stats(util);
static const uint stattot = 256;
uint statcnt = stats.m_stats.size();
require(statcnt == 1);
{
const UtilStat* stat = stats.find("root");
require(stat != 0);
require(stat->m_id == 0);
}
for (uint i = 1; i < stattot; i++)
{
UtilName name("test", i);
uint parent = myrandom(statcnt);
uint flags = 0;
const UtilStat* stat = stats.create(name, parent, flags);
require(stat != 0);
require(stat->m_id == statcnt);
require(strcmp(stat->m_name, name) == 0);
statcnt++;
require(statcnt == stats.m_stats.size());
}
require(statcnt == stattot);
for (uint k = 0; k < 10 * stattot; k++)
{
uint i = myrandom(statcnt);
require(i < statcnt);
UtilStat* stat = stats.get(i);
require(stat != 0);
if (i == 0)
{
UtilName name("root");
require(strcmp(stat->m_name, name) == 0);
UtilStat* stat2 = stats.find(name);
require(stat == stat2);
}
else
{
UtilName name("test", i);
require(strcmp(stat->m_name, name) == 0);
UtilStat* stat2 = stats.find(name);
require(stat == stat2);
uint v =myrandom();
stat->add(v);
}
}
// iter
bool iterseen[stattot];
for (uint i = 0; i < stattot; i++)
iterseen[i] = false;
// root is skipped
const UtilStat* stat = stats.next(0);
iterseen[0] = true;
uint itercnt = 1;
while (stat != 0)
{
itercnt++;
require(itercnt <= statcnt);
const uint i = stat->m_id;
require(i < stattot);
require(iterseen[i] == false);
iterseen[i] = true;
for (uint j = i; j != 0; )
{
const UtilStat* stat = stats.get(j);
require(stat != 0);
require(iterseen[j] == true);
ndbout << j;
j = stat->m_parent;
if (j != 0)
ndbout << " ";
else
ndbout << endl;
}
stat = stats.next(i);
}
for (uint i = 0; i < stattot; i++)
require(iterseen[i] == true);
require(itercnt == statcnt);
return 0;
}
static void
myseed()
{
const char* p = NdbEnv_GetEnv("TEST_NDBIMPORTUTIL_SEED", (char*)0, 0);
unsigned seed = p != 0 ? (unsigned)atoi(p) : (uint)NdbHost_GetProcessId();
ndbout << "seed=" << seed << endl;
ndb_srand(seed);
}
static bool
mycase(const char* name)
{
const char* p = NdbEnv_GetEnv("TEST_NDBIMPORTUTIL_CASE", (char*)0, 0);
return p == 0 || strcmp(p, name) == 0;
}
static int
testmain()
{
ndb_init();
#ifdef VM_TRACE
signal(SIGABRT, SIG_DFL);
signal(SIGSEGV, SIG_DFL);
#endif
mybigtest =
#ifdef VM_TRACE
false;
#else
true;
#endif
myseed();
if (mycase("testlist") && testlist() != 0)
return -1;
if (mycase("testrowlist1") && testrowlist1() != 0)
return -1;
if (mycase("testrowlist2") && testrowlist2() != 0)
return -1;
if (mycase("testrowmap1") && testrowmap1() != 0)
return -1;
if (mycase("testrowmap2") && testrowmap2() != 0)
return -1;
if (mycase("testbuf") && testbuf() != 0)
return -1;
if (mycase("testfile") && testfile() != 0)
return -1;
if (mycase("testprint") && testprint() != 0)
return -1;
if (mycase("teststat") && teststat() != 0)
return -1;
struct ndb_rusage ru;
require(Ndb_GetRUsage(&ru, false) == 0);
ndbout << "utime=" << ru.ru_utime/1000
<< " stime=" << ru.ru_stime/1000 << " (ms)" << endl;
return 0;
}
TAPTEST(NdbImportUtil)
{
int ret = testmain();
return (ret == 0);
}
#endif
Reference in New Issue View Git Blame Copy Permalink