summaryrefslogtreecommitdiff
path: root/lib/output/pq
diff options
context:
space:
mode:
Diffstat (limited to 'lib/output/pq')
-rw-r--r--lib/output/pq/updateDatabase.cpp116
-rw-r--r--lib/output/pq/updateDatabase.h1
2 files changed, 117 insertions, 0 deletions
diff --git a/lib/output/pq/updateDatabase.cpp b/lib/output/pq/updateDatabase.cpp
index 3829da9..457f744 100644
--- a/lib/output/pq/updateDatabase.cpp
+++ b/lib/output/pq/updateDatabase.cpp
@@ -7,6 +7,7 @@
#include <eventSourceBase.h>
#include <helpers.h>
#include <input/mysqlConn.h>
+#include <input/mysqlRecordSet.h>
#include <input/replStream.h>
#include <input/sql/selectColumns.h>
#include <input/sql/showMasterStatus.h>
@@ -92,4 +93,119 @@ namespace MyGrate::Output::Pq {
});
tables.emplace(tableName, std::move(tableDef));
}
+
+ struct WritePqCopyStream {
+ ~WritePqCopyStream()
+ {
+ fputc('\n', out);
+ }
+
+ void
+ nextField()
+ {
+ fputc('\t', out);
+ }
+
+ void operator()(std::nullptr_t) const
+ {
+ fputs("\\N", out);
+ }
+#define BASIC_PRINT(T, fmt) \
+ void operator()(T v) const \
+ { \
+ fprintf(out, fmt, v); \
+ }
+ BASIC_PRINT(double, "%f")
+ BASIC_PRINT(float, "%f")
+ BASIC_PRINT(int8_t, "%hhd")
+ BASIC_PRINT(uint8_t, "%hhu")
+ BASIC_PRINT(int16_t, "%hd")
+ BASIC_PRINT(uint16_t, "%hu")
+ BASIC_PRINT(int32_t, "%d")
+ BASIC_PRINT(uint32_t, "%u")
+ BASIC_PRINT(int64_t, "%ld")
+ BASIC_PRINT(uint64_t, "%lu")
+#undef BASIC_PRINT
+ void operator()(timespec) const
+ {
+ throw std::logic_error("timespec not implemented");
+ }
+ void
+ operator()(Date v) const
+ {
+ fprintf(out, "%d-%d-%d", v.year, v.month, v.day);
+ }
+ void
+ operator()(Time v) const
+ {
+ fprintf(out, "%d:%d:%d", v.hour, v.minute, v.second);
+ }
+ void
+ operator()(DateTime v) const
+ {
+ operator()(static_cast<Date &>(v));
+ fputc('T', out);
+ operator()(static_cast<Time &>(v));
+ }
+ void
+ operator()(std::string_view v) const
+ {
+ auto pos {v.begin()};
+ while (pos != v.end()) {
+ auto esc = std::find_if(pos, v.end(), [](unsigned char c) {
+ return std::iscntrl(c);
+ });
+ if (esc != pos) {
+ fwrite(pos, esc - pos, 1, out);
+ pos = esc;
+ }
+ while (pos != v.end()) {
+ fprintf(out, "\\%03o", *pos);
+ pos++;
+ }
+ }
+ }
+ void operator()(BitSet) const
+ {
+ throw std::logic_error("bitset not implemented");
+ }
+ void
+ operator()(Blob v) const
+ {
+ fputs("\\\\x", out);
+ std::for_each(v.begin(), v.end(), [this](auto b) {
+ fprintf(out, "%02hhx", (uint8_t)b);
+ });
+ }
+
+ FILE * out;
+ };
+
+ void
+ UpdateDatabase::copyTableContent(Input::MySQLConn * conn, const char * table)
+ {
+ auto out = beginBulkUpload(schema.c_str(), table);
+ std::stringstream sf;
+ unsigned int ordinal {0};
+ for (const auto & col : tables.at(table)->columns) {
+ scprintf<"%? %?">(sf, !ordinal++ ? "SELECT " : ", ", col->name);
+ }
+ sf << " FROM " << table;
+ auto stmt {conn->prepare(sf.str().c_str(), 0)};
+ stmt->execute({});
+ auto sourceCursor {stmt->cursor()};
+
+ const auto cols = sourceCursor->columns();
+ while (sourceCursor->fetch()) {
+ WritePqCopyStream cs {out};
+ for (auto ordinal {0U}; ordinal < cols; ordinal += 1) {
+ if (ordinal) {
+ cs.nextField();
+ }
+ sourceCursor->at(ordinal).visit(cs);
+ }
+ }
+
+ fclose(out);
+ }
}
diff --git a/lib/output/pq/updateDatabase.h b/lib/output/pq/updateDatabase.h
index e5b06ef..20ea96a 100644
--- a/lib/output/pq/updateDatabase.h
+++ b/lib/output/pq/updateDatabase.h
@@ -42,6 +42,7 @@ namespace MyGrate::Output::Pq {
const char * db, int sid, const char * sc);
void addTable(Input::MySQLConn *, const char * tableName);
+ void copyTableContent(Input::MySQLConn *, const char * tableName);
const uint64_t source;
const std::string schema;