github-starred/ultimatepp
2
0
Fork 0
mirror of https://github.com/ultimatepp/ultimatepp.git synced 2026-08-24 14:22:40 -06:00
Code Issues 33 Projects Packages Wiki Activity Actions

Sql: MassInsert optimized for PGSQL/MySql

Browse source 
git-svn-id: svn://ultimatepp.org/upp/trunk@5352 f0d560ea-af0d-0410-9eb7-867de7ffcac7
This commit is contained in:
cxl 2012-09-11 18:29:00 +00:00
parent fb53814a17
commit dda4dbff56
1 changed files with 134 additions and 110 deletions
Download patch file Download diff file Expand all files Collapse all files

244
uppsrc/Sql/MassInsert.cpp
View file

@ -1,110 +1,134 @@
#include "Sql.h"
NAMESPACE_UPP
SqlMassInsert::~SqlMassInsert()
{
Flush();
}
SqlMassInsert& SqlMassInsert::operator()(SqlId col, const Value& val)
{
if(pos == 0) {
cache.Add().nulls = 0;
cache.Top();
}
if(cache.GetCount() == 1)
column.Add(~col);
else
ASSERT(column[pos] == col.Quoted() || column[pos] == ~col);
Row& r = cache.Top();
r.value.Add(val);
if(IsNull(val))
r.nulls |= (1 << pos);
pos++;
ASSERT(pos < 30);
return *this;
}
SqlMassInsert& SqlMassInsert::EndRow(SqlBool remove)
{
cache.Top().remove = remove;
if(cache.GetCount() && cache[0].value.GetCount() * cache.GetCount() > 5000)
Flush();
ASSERT(column.GetCount() == pos);
pos = 0;
return *this;
}
void SqlMassInsert::Flush()
{
const dword DONE = 0xffffffff;
if(cache.GetCount() == 0)
return;
if(use_transaction)
sql.GetSession().Begin();
SqlBool remove;
bool doremove = false;
for(int ii = 0; ii < cache.GetCount(); ii++) {
SqlBool rm = cache[ii].remove;
if(!rm.IsEmpty()) {
doremove = true;
remove = remove || rm;
}
}
if(doremove)
sql * Delete(table).Where(remove);
for(int ii = 0; ii < cache.GetCount(); ii++) {
dword nulls = cache[ii].nulls;
if(nulls != DONE) {
String insert;
insert << "insert into " + ~table + '(';
bool nextcol = false;
for(int i = 0; i < column.GetCount(); i++) {
if(!(nulls & (1 << i))) {
if(nextcol)
insert << ", ";
nextcol = true;
insert << column[i];
}
}
insert << ')';
bool nextsel = false;
for(int i = ii; i < cache.GetCount(); i++) {
Row& r = cache[i];
if(r.nulls == nulls) {
r.nulls = DONE;
if(nextsel)
insert << " union all";
nextsel = true;
insert << " select ";
bool nextval = false;
for(int i = 0; i < r.value.GetCount(); i++)
if(!(nulls & (1 << i))) {
if(nextval)
insert << ", ";
nextval = true;
insert << SqlCompile(sql.GetDialect(), SqlFormat(r.value[i]));
}
if(sql.GetDialect() == ORACLE)
insert << " from dual";
}
}
sql.Execute(insert);
}
}
if(sql.WasError()) {
error = true;
if(use_transaction)
sql.GetSession().Rollback();
}
else
if(use_transaction)
sql.GetSession().Commit();
cache.Clear();
column.Clear();
pos = 0;
}
END_UPP_NAMESPACE
#include "Sql.h"
NAMESPACE_UPP
SqlMassInsert::~SqlMassInsert()
{
Flush();
}
SqlMassInsert& SqlMassInsert::operator()(SqlId col, const Value& val)
{
if(pos == 0) {
cache.Add().nulls = 0;
cache.Top();
}
if(cache.GetCount() == 1)
column.Add(~col);
else
ASSERT(column[pos] == col.Quoted() || column[pos] == ~col);
Row& r = cache.Top();
r.value.Add(val);
if(IsNull(val))
r.nulls |= (1 << pos);
pos++;
ASSERT(pos < 30);
return *this;
}
SqlMassInsert& SqlMassInsert::EndRow(SqlBool remove)
{
cache.Top().remove = remove;
if(cache.GetCount() && cache[0].value.GetCount() * cache.GetCount() > 5000)
Flush();
ASSERT(column.GetCount() == pos);
pos = 0;
return *this;
}
void SqlMassInsert::Flush()
{
const dword DONE = 0xffffffff;
if(cache.GetCount() == 0)
return;
if(use_transaction)
sql.GetSession().Begin();
SqlBool remove;
bool doremove = false;
for(int ii = 0; ii < cache.GetCount(); ii++) {
SqlBool rm = cache[ii].remove;
if(!rm.IsEmpty()) {
doremove = true;
remove = remove || rm;
}
}
if(doremove)
sql * Delete(table).Where(remove);
String insert;
int dialect = sql.GetDialect();
if(dialect == MY_SQL || dialect == PGSQL) {
insert << "insert into " + ~table + '(';
for(int i = 0; i < column.GetCount(); i++) {
if(i)
insert << ", ";
insert << column[i];
}
insert << ") values ";
for(int i = 0; i < cache.GetCount(); i++) {
Row& r = cache[i];
if(i)
insert << ", ";
insert << "(";
bool nextval = false;
for(int i = 0; i < r.value.GetCount(); i++) {
if(i)
insert << ", ";
insert << SqlCompile(dialect, SqlFormat(r.value[i]));
}
insert << ")";
}
}
else
for(int ii = 0; ii < cache.GetCount(); ii++) {
dword nulls = cache[ii].nulls;
if(nulls != DONE) {
insert << "insert into " + ~table + '(';
bool nextcol = false;
for(int i = 0; i < column.GetCount(); i++) {
if(!(nulls & (1 << i))) {
if(nextcol)
insert << ", ";
nextcol = true;
insert << column[i];
}
}
insert << ')';
bool nextsel = false;
for(int i = ii; i < cache.GetCount(); i++) {
Row& r = cache[i];
if(r.nulls == nulls) {
r.nulls = DONE;
if(nextsel)
insert << " union all";
nextsel = true;
insert << " select ";
bool nextval = false;
for(int i = 0; i < r.value.GetCount(); i++)
if(!(nulls & (1 << i))) {
if(nextval)
insert << ", ";
nextval = true;
insert << SqlCompile(dialect, SqlFormat(r.value[i]));
}
if(dialect == ORACLE)
insert << " from dual";
}
}
}
}
sql.Execute(insert);
if(sql.WasError()) {
error = true;
if(use_transaction)
sql.GetSession().Rollback();
}
else
if(use_transaction)
sql.GetSession().Commit();
cache.Clear();
column.Clear();
pos = 0;
}
END_UPP_NAMESPACE