♻️Remove Cassandra Support
This commit is contained in:
Anton SHEPILOV
committed by
Anton Shepilov
Anton Shepilov
parent
8214a5f6d4
commit
8f75c04fdd
@@ -29,14 +29,6 @@ export default function printConfigSummary(useIcons: boolean = !!process.env.HAV
|
||||
"user",
|
||||
user,
|
||||
],
|
||||
cassandra: ({ contactPoints, localDataCenter, keyspace }) => [
|
||||
"host",
|
||||
contactPoints,
|
||||
"datacenter",
|
||||
localDataCenter,
|
||||
"keyspace",
|
||||
keyspace,
|
||||
],
|
||||
S3: ({ endPoint, port, accessKey, bucket }) => [
|
||||
"host",
|
||||
[endPoint, port],
|
||||
@@ -62,7 +54,6 @@ export default function printConfigSummary(useIcons: boolean = !!process.env.HAV
|
||||
elasticsearch: "ES",
|
||||
opensearch: "OS",
|
||||
mongodb: "mongo",
|
||||
cassandra: "Cass.",
|
||||
};
|
||||
if (useIcons) {
|
||||
const makeLyingLengthIcon = str => ({ length: 1, toString: _ => str } as string);
|
||||
|
||||
@@ -1,2 +0,0 @@
|
||||
node_modules
|
||||
yarn.lock
|
||||
@@ -1,4 +0,0 @@
|
||||
Edit the file to your needs
|
||||
|
||||
yarn install
|
||||
node index.js
|
||||
@@ -1,214 +0,0 @@
|
||||
/* eslint-disable */
|
||||
var cassandra = require("cassandra-driver");
|
||||
|
||||
var fromAuthProvider = new cassandra.auth.PlainTextAuthProvider("", "");
|
||||
var fromContactPoints = [""];
|
||||
var fromKeyspace = "tdrive";
|
||||
|
||||
var toAuthProvider = new cassandra.auth.PlainTextAuthProvider("", "");
|
||||
var toContactPoints = [""];
|
||||
var toKeyspace = "tdrive";
|
||||
|
||||
var forceUpdateAll = false;
|
||||
var ignoreTables = ["notification"];
|
||||
var countersTables = ["statistics", "stats_counter", "channel_counters", "scheduled_queue_counter"];
|
||||
|
||||
var specialConversions = {
|
||||
"tdrive.group_user.date_added": value => {
|
||||
return new Date(value).getTime();
|
||||
},
|
||||
};
|
||||
|
||||
// -- start process
|
||||
|
||||
var fromClient = new cassandra.Client({
|
||||
localDataCenter: "datacenter1",
|
||||
contactPoints: fromContactPoints,
|
||||
authProvider: fromAuthProvider,
|
||||
keyspace: fromKeyspace,
|
||||
});
|
||||
|
||||
var toClient = new cassandra.Client({
|
||||
localDataCenter: "datacenter1",
|
||||
contactPoints: toContactPoints,
|
||||
authProvider: toAuthProvider,
|
||||
keyspace: toKeyspace,
|
||||
queryOptions: {
|
||||
consistency: cassandra.types.consistencies.quorum,
|
||||
},
|
||||
});
|
||||
|
||||
// -- Get all tables and copy schema
|
||||
|
||||
async function client(origin, query, parameters, options) {
|
||||
return await new Promise((resolve, reject) => {
|
||||
origin.execute(query, [], {}, function (err, result) {
|
||||
if (err) {
|
||||
reject({ err, result });
|
||||
} else {
|
||||
resolve(result);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
(async () => {
|
||||
const result = await client(
|
||||
fromClient,
|
||||
"SELECT table_name from system_schema.tables WHERE keyspace_name = '" + fromKeyspace + "'",
|
||||
[],
|
||||
{},
|
||||
);
|
||||
for (row of result.rows) {
|
||||
try {
|
||||
const fromTable = fromKeyspace + "." + row.table_name;
|
||||
const toTable = toKeyspace + "." + row.table_name;
|
||||
|
||||
if (ignoreTables.includes(row.table_name)) {
|
||||
console.log(fromTable.padEnd(50) + " | " + "ignored".padEnd(20) + " | ⏺");
|
||||
continue;
|
||||
}
|
||||
|
||||
let fromCount = 0;
|
||||
|
||||
const destColumns = (
|
||||
await client(
|
||||
toClient,
|
||||
"SELECT column_name from system_schema.columns where keyspace_name = '" +
|
||||
toKeyspace +
|
||||
"' and table_name = '" +
|
||||
row.table_name +
|
||||
"'",
|
||||
[],
|
||||
{},
|
||||
)
|
||||
).rows.map(r => r.column_name);
|
||||
|
||||
try {
|
||||
const fromResult = await client(
|
||||
fromClient,
|
||||
"SELECT count(*) from " + fromTable + "",
|
||||
[],
|
||||
{},
|
||||
);
|
||||
fromCount = parseInt(fromResult.rows[0].count);
|
||||
} catch (err) {
|
||||
fromCount = NaN;
|
||||
}
|
||||
|
||||
if (destColumns.length === 0) {
|
||||
if (fromCount > 0)
|
||||
console.log(
|
||||
fromTable.padEnd(50) + " | " + ("not_in_dest" + "/" + fromCount).padEnd(20) + " | ⏺",
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
const toResult = await client(toClient, "SELECT count(*) from " + toTable + "", [], {});
|
||||
const toCount = parseInt(toResult.rows[0].count);
|
||||
|
||||
console.log(
|
||||
fromTable.padEnd(50) +
|
||||
" | " +
|
||||
(toCount + "/" + fromCount).padEnd(20) +
|
||||
" | " +
|
||||
(toCount >= fromCount ? "✅" : "❌"),
|
||||
);
|
||||
|
||||
if (row.table_name.indexOf("counter") >= 0 || countersTables.includes(row.table_name)) {
|
||||
console.log(
|
||||
fromTable.padEnd(50) + " | " + ("counter_table" + "/" + fromCount).padEnd(20) + " | 🧮",
|
||||
);
|
||||
if (fromCount > toCount || !fromCount || forceUpdateAll) {
|
||||
await new Promise(r => {
|
||||
fromClient.eachRow(
|
||||
"SELECT JSON * from " + fromTable,
|
||||
[],
|
||||
{ prepare: true, fetchSize: 1000 },
|
||||
async function (n, row) {
|
||||
//
|
||||
//TODO handle counters (it is special !)
|
||||
},
|
||||
async function (err, result) {
|
||||
if (result && result.nextPage) {
|
||||
await new Promise(r => setTimeout(r, 1000));
|
||||
result.nextPage();
|
||||
} else {
|
||||
r();
|
||||
}
|
||||
},
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
if (fromCount > toCount || !fromCount || forceUpdateAll) {
|
||||
await new Promise(r => {
|
||||
fromClient.eachRow(
|
||||
"SELECT JSON * from " + fromTable,
|
||||
[],
|
||||
{ prepare: true, fetchSize: 1000 },
|
||||
async function (n, row) {
|
||||
try {
|
||||
const json = JSON.parse(row["[json]"]);
|
||||
|
||||
//The from table can have additional depreciated fields, we need to remove them
|
||||
const filteredJson = {};
|
||||
for (const col of destColumns) {
|
||||
if (
|
||||
typeof json[col] == "string" &&
|
||||
(json[col] || "").match(
|
||||
/[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}.[0-9]+/,
|
||||
)
|
||||
) {
|
||||
json[col] = json[col].split(".")[0];
|
||||
}
|
||||
|
||||
if (specialConversions[fromTable + "." + col]) {
|
||||
json[col] = specialConversions[fromTable + "." + col](json[col]);
|
||||
}
|
||||
|
||||
if (json[col] !== undefined) filteredJson[col] = json[col];
|
||||
}
|
||||
|
||||
await client(
|
||||
toClient,
|
||||
"INSERT INTO " +
|
||||
toTable +
|
||||
" JSON '" +
|
||||
JSON.stringify(filteredJson).replace(/'/g, "'$&") +
|
||||
"'",
|
||||
[],
|
||||
{},
|
||||
);
|
||||
} catch (err) {
|
||||
console.log(err);
|
||||
}
|
||||
},
|
||||
async function (err, result) {
|
||||
if (result && result.nextPage) {
|
||||
await new Promise(r => setTimeout(r, 1000));
|
||||
result.nextPage();
|
||||
} else {
|
||||
r();
|
||||
}
|
||||
},
|
||||
);
|
||||
});
|
||||
}
|
||||
} catch (err) {
|
||||
console.log(
|
||||
fromTable.padEnd(50) + " | " + ("error" + "/" + fromCount).padEnd(20) + " | ❌",
|
||||
);
|
||||
}
|
||||
} catch (err) {
|
||||
console.log(err);
|
||||
continue;
|
||||
}
|
||||
|
||||
//TODO copy content
|
||||
}
|
||||
})();
|
||||
@@ -1,9 +0,0 @@
|
||||
{
|
||||
"name": "copy-cluster",
|
||||
"version": "1.0.0",
|
||||
"main": "index.js",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"cassandra-driver": "^4.6.3"
|
||||
}
|
||||
}
|
||||
@@ -1,4 +0,0 @@
|
||||
Edit the file to your needs
|
||||
|
||||
yarn install
|
||||
node index.js
|
||||
@@ -1,88 +0,0 @@
|
||||
/* eslint-disable */
|
||||
var cassandra = require("cassandra-driver");
|
||||
|
||||
var authProvider = new cassandra.auth.PlainTextAuthProvider("", "");
|
||||
var contactPoints = [""];
|
||||
var from = "table_a";
|
||||
var to = "table_b";
|
||||
|
||||
var client = new cassandra.Client({
|
||||
localDataCenter: "datacenter1",
|
||||
contactPoints: contactPoints,
|
||||
authProvider: authProvider,
|
||||
keyspace: "tdrive",
|
||||
});
|
||||
|
||||
//Copy object to other table
|
||||
(async () => {
|
||||
const query = "SELECT * FROM " + from;
|
||||
const parameters = [];
|
||||
|
||||
let count = 0;
|
||||
let max = 0;
|
||||
let min = new Date().getTime();
|
||||
|
||||
let rows = [];
|
||||
|
||||
const options = { prepare: true, fetchSize: 1000 };
|
||||
|
||||
await new Promise(r => {
|
||||
client.eachRow(
|
||||
query,
|
||||
parameters,
|
||||
options,
|
||||
function (n, row) {
|
||||
min = Math.min(parseInt(row.created_at.toString()), min);
|
||||
max = Math.max(parseInt(row.created_at.toString()), max);
|
||||
count++;
|
||||
if (count % 100 == 0) {
|
||||
console.log(count);
|
||||
}
|
||||
rows.push(row);
|
||||
},
|
||||
function (err, result) {
|
||||
if (result && result.nextPage) {
|
||||
result.nextPage();
|
||||
} else {
|
||||
r();
|
||||
}
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
console.log("Downloaded ", count, " rows of ", from, ". Starting copy...");
|
||||
|
||||
let copyCount = 0;
|
||||
|
||||
for (let i = 0; i < count; i++) {
|
||||
copyCount++;
|
||||
if (copyCount % 100 == 0) {
|
||||
console.log(copyCount, "of", count);
|
||||
}
|
||||
|
||||
const row = rows[i];
|
||||
|
||||
await new Promise(r => {
|
||||
let query =
|
||||
"UPDATE " +
|
||||
to +
|
||||
" SET answers=?, created_at=?, created_by=?, last_activity=?, participants=?, updated_at=? WHERE id=?";
|
||||
client.execute(
|
||||
query,
|
||||
[
|
||||
row.answers,
|
||||
row.created_at,
|
||||
row.created_by,
|
||||
row.last_activity,
|
||||
row.participants,
|
||||
row.updated_at,
|
||||
row.id,
|
||||
],
|
||||
() => {
|
||||
r();
|
||||
},
|
||||
);
|
||||
});
|
||||
}
|
||||
console.log("Ended with ", count, "threads");
|
||||
})();
|
||||
@@ -1,6 +0,0 @@
|
||||
{
|
||||
"name": "copy-table",
|
||||
"version": "1.0.0",
|
||||
"main": "index.js",
|
||||
"license": "MIT"
|
||||
}
|
||||
Reference in New Issue
Block a user