smartclickhouse/ts/smartclickhouse.classes.timedatatable.ts

305 lines
9.9 KiB
TypeScript
Raw Normal View History

2022-03-14 13:29:23 +00:00
import * as plugins from './smartclickhouse.plugins.js';
import { SmartClickHouseDb } from './smartclickhouse.classes.smartclickhouse.js';
2022-03-02 15:35:20 +00:00
2022-08-05 11:31:11 +00:00
export type TClickhouseColumnDataType =
| 'String'
| "DateTime64(3, 'Europe/Berlin')"
| 'Float64'
| 'Array(String)'
| 'Array(Float64)';
2022-03-07 14:49:47 +00:00
export interface IColumnInfo {
database: string;
table: string;
name: string;
type: TClickhouseColumnDataType;
position: string;
default_kind: string;
default_expression: string;
data_compressed_bytes: string;
data_uncompressed_bytes: string;
marks_bytes: string;
comment: string;
is_in_partition_key: 0 | 1;
is_in_sorting_key: 0 | 1;
is_in_primary_key: 0 | 1;
is_in_sampling_key: 0 | 1;
compression_codec: string;
character_octet_length: null;
numeric_precision: null;
numeric_precision_radix: null;
numeric_scale: null;
datetime_precision: '3';
}
export interface ITimeDataTableOptions {
tableName: string;
retainDataForDays: number;
}
2022-03-02 15:35:20 +00:00
export class TimeDataTable {
2022-03-07 14:49:47 +00:00
public static async getTable(smartClickHouseDbRefArg: SmartClickHouseDb, tableNameArg: string) {
const newTable = new TimeDataTable(smartClickHouseDbRefArg, {
tableName: tableNameArg,
2022-08-05 11:31:11 +00:00
retainDataForDays: 30,
});
2022-03-02 15:35:20 +00:00
2022-07-27 21:11:03 +00:00
await newTable.setup();
return newTable;
}
// INSTANCE
public healingDeferred: plugins.smartpromise.Deferred<any>;
public smartClickHouseDbRef: SmartClickHouseDb;
public options: ITimeDataTableOptions;
constructor(smartClickHouseDbRefArg: SmartClickHouseDb, optionsArg: ITimeDataTableOptions) {
this.smartClickHouseDbRef = smartClickHouseDbRefArg;
this.options = optionsArg;
}
public async setup() {
2022-03-02 15:35:20 +00:00
// create table in clickhouse
2024-06-14 14:33:00 +00:00
await this.smartClickHouseDbRef.clickhouseHttpClient.queryPromise(`
2022-07-28 14:53:07 +00:00
CREATE TABLE IF NOT EXISTS ${this.smartClickHouseDbRef.options.database}.${this.options.tableName} (
timestamp DateTime64(3, 'Europe/Berlin'),
message String
2022-03-14 12:52:42 +00:00
) ENGINE=MergeTree() ORDER BY timestamp`);
2022-08-05 11:31:11 +00:00
// lets adjust the TTL
2024-06-14 14:33:00 +00:00
await this.smartClickHouseDbRef.clickhouseHttpClient.queryPromise(`
2022-07-28 14:53:07 +00:00
ALTER TABLE ${this.smartClickHouseDbRef.options.database}.${this.options.tableName} MODIFY TTL toDateTime(timestamp) + INTERVAL ${this.options.retainDataForDays} DAY
`);
2022-03-07 14:49:47 +00:00
2022-07-27 21:11:03 +00:00
await this.updateColumns();
2022-08-05 11:31:11 +00:00
console.log(`=======================`);
2022-03-07 14:49:47 +00:00
console.log(
2022-07-27 21:11:03 +00:00
`table with name "${this.options.tableName}" in database ${this.smartClickHouseDbRef.options.database} has the following columns:`
2022-03-07 14:49:47 +00:00
);
2022-07-27 21:11:03 +00:00
for (const column of this.columns) {
2022-03-07 14:49:47 +00:00
console.log(`>> ${column.name}: ${column.type}`);
}
console.log('^^^^^^^^^^^^^^\n');
2022-03-02 15:35:20 +00:00
}
2022-03-02 15:44:09 +00:00
2022-03-07 14:49:47 +00:00
public columns: IColumnInfo[] = [];
/**
* updates the columns
*/
public async updateColumns() {
2024-06-14 14:33:00 +00:00
this.columns = await this.smartClickHouseDbRef.clickhouseHttpClient.queryPromise(`
2022-03-07 14:49:47 +00:00
SELECT * FROM system.columns
WHERE database LIKE '${this.smartClickHouseDbRef.options.database}'
2022-07-28 14:53:07 +00:00
AND table LIKE '${this.options.tableName}' FORMAT JSONEachRow
2022-03-07 14:49:47 +00:00
`);
return this.columns;
}
2022-03-02 15:44:09 +00:00
/**
* stores a json and tries to map it to the nested syntax
*/
2022-03-07 14:49:47 +00:00
public async addData(dataArg: any) {
2022-08-01 10:52:53 +00:00
if (this.healingDeferred) {
return;
}
2022-07-27 20:42:08 +00:00
2022-03-07 14:49:47 +00:00
// the storageJson
let storageJson: { [key: string]: any } = {};
2022-08-05 11:31:11 +00:00
2022-03-07 14:49:47 +00:00
// helper stuff
2022-08-05 11:31:11 +00:00
2022-03-08 14:12:51 +00:00
const getClickhouseTypeForValue = (valueArg: any): TClickhouseColumnDataType => {
2022-08-05 11:31:11 +00:00
const typeConversion: { [key: string]: TClickhouseColumnDataType } = {
2022-03-08 14:12:51 +00:00
string: 'String',
number: 'Float64',
undefined: null,
2022-08-05 11:31:11 +00:00
null: null,
2022-03-08 14:12:51 +00:00
};
if (valueArg instanceof Array) {
2022-08-05 11:31:11 +00:00
const arrayType = typeConversion[typeof valueArg[0] as string];
2022-03-08 14:12:51 +00:00
if (!arrayType) {
return null;
} else {
return `Array(${arrayType})` as TClickhouseColumnDataType;
}
}
2022-08-05 11:31:11 +00:00
return typeConversion[typeof valueArg as string];
};
const checkPath = async (
pathArg: string,
typeArg: TClickhouseColumnDataType,
prechecked = false
) => {
2022-03-07 14:49:47 +00:00
let columnFound = false;
for (const column of this.columns) {
if (pathArg === column.name) {
columnFound = true;
break;
}
}
if (!columnFound) {
2022-03-08 14:12:51 +00:00
if (!prechecked) {
await this.updateColumns();
await checkPath(pathArg, typeArg, true);
return;
}
2022-08-05 11:31:11 +00:00
const alterString = `ALTER TABLE ${this.smartClickHouseDbRef.options.database}.${this.options.tableName} ADD COLUMN ${pathArg} ${typeArg} FIRST`;
2022-03-08 14:12:51 +00:00
try {
2024-06-14 14:33:00 +00:00
await this.smartClickHouseDbRef.clickhouseHttpClient.queryPromise(`
2022-03-08 14:12:51 +00:00
${alterString}
2022-03-07 14:49:47 +00:00
`);
2022-08-05 11:31:11 +00:00
} catch (err) {
2022-03-08 14:12:51 +00:00
console.log(alterString);
for (const column of this.columns) {
console.log(column.name);
}
}
2022-03-07 14:49:47 +00:00
await this.updateColumns();
}
};
// key checking
const flatDataArg = plugins.smartobject.toFlatObject(dataArg);
for (const key of Object.keys(flatDataArg)) {
const value = flatDataArg[key];
if (key === 'timestamp' && typeof value !== 'number') {
throw new Error('timestamp must be of type number');
} else if (key === 'timestamp') {
storageJson.timestamp = flatDataArg[key];
continue;
}
// lets deal with the rest
const clickhouseType = getClickhouseTypeForValue(value);
2022-03-08 14:12:51 +00:00
if (!clickhouseType) {
continue;
}
2022-03-07 14:49:47 +00:00
await checkPath(key, clickhouseType);
storageJson[key] = value;
}
2024-06-14 14:33:00 +00:00
const result = await this.smartClickHouseDbRef.clickhouseHttpClient
2022-08-05 11:31:11 +00:00
.insertPromise(this.smartClickHouseDbRef.options.database, this.options.tableName, [
storageJson,
])
.catch(async () => {
if (this.healingDeferred) {
return;
}
this.healingDeferred = plugins.smartpromise.defer();
console.log(`Ran into an error. Trying to set up things properly again.`);
await this.smartClickHouseDbRef.pingDatabaseUntilAvailable();
await this.smartClickHouseDbRef.createDatabase();
await this.setup();
this.columns = [];
this.healingDeferred.resolve();
this.healingDeferred = null;
});
2022-03-07 14:49:47 +00:00
return result;
}
2024-06-14 14:33:00 +00:00
/**
* deletes the entire table
*/
public async delete() {
await this.smartClickHouseDbRef.clickhouseHttpClient.queryPromise(`
DROP TABLE IF EXISTS ${this.smartClickHouseDbRef.options.database}.${this.options.tableName}
`);
this.columns = [];
}
/**
* deletes entries older than a specified number of days
* @param days number of days
*/
public async deleteOldEntries(days: number) {
2024-06-14 14:56:39 +00:00
// Perform the deletion operation
2024-06-14 14:33:00 +00:00
await this.smartClickHouseDbRef.clickhouseHttpClient.queryPromise(`
2024-06-14 14:56:39 +00:00
ALTER TABLE ${this.smartClickHouseDbRef.options.database}.${this.options.tableName}
DELETE WHERE timestamp < now() - INTERVAL ${days} DAY
`);
await this.waitForMutations();
}
public async waitForMutations() {
// Wait for the mutation to complete
let mutations;
do {
mutations = await this.smartClickHouseDbRef.clickhouseHttpClient.queryPromise(`
SELECT count() AS mutations_count FROM system.mutations
WHERE is_done = 0 AND table = '${this.options.tableName}'
2024-06-14 14:33:00 +00:00
`);
2024-06-14 14:56:39 +00:00
if (mutations[0] && mutations[0].mutations_count > 0) {
console.log('Waiting for mutations to complete...');
await new Promise((resolve) => setTimeout(resolve, 1000));
}
} while (mutations[0] && mutations[0].mutations_count > 0);
2024-06-14 14:33:00 +00:00
}
public async getLastEntries(count: number) {
const result = await this.smartClickHouseDbRef.clickhouseHttpClient.queryPromise(`
SELECT * FROM ${this.smartClickHouseDbRef.options.database}.${this.options.tableName}
ORDER BY timestamp DESC
LIMIT ${count} FORMAT JSONEachRow
`);
return result;
}
public async getEntriesNewerThan(unixTimestamp: number) {
const result = await this.smartClickHouseDbRef.clickhouseHttpClient.queryPromise(`
SELECT * FROM ${this.smartClickHouseDbRef.options.database}.${this.options.tableName}
WHERE timestamp > toDateTime(${unixTimestamp / 1000}) FORMAT JSONEachRow
`);
return result;
}
public async getEntriesBetween(unixTimestampStart: number, unixTimestampEnd: number) {
const result = await this.smartClickHouseDbRef.clickhouseHttpClient.queryPromise(`
SELECT * FROM ${this.smartClickHouseDbRef.options.database}.${this.options.tableName}
WHERE timestamp > toDateTime(${unixTimestampStart / 1000})
AND timestamp < toDateTime(${unixTimestampEnd / 1000}) FORMAT JSONEachRow
`);
return result;
}
/**
* streams all new entries using an observable
*/
public streamNewEntries(): plugins.smartrx.rxjs.Observable<any> {
return new plugins.smartrx.rxjs.Observable((observer) => {
const pollInterval = 1000; // Poll every 1 second
let lastTimestamp: number;
const fetchLastEntryTimestamp = async () => {
const lastEntry = await this.smartClickHouseDbRef.clickhouseHttpClient.queryPromise(`
SELECT max(timestamp) as lastTimestamp FROM ${this.smartClickHouseDbRef.options.database}.${this.options.tableName} FORMAT JSONEachRow
`);
lastTimestamp = lastEntry.length
? new Date(lastEntry[0].lastTimestamp).getTime()
: Date.now();
};
const fetchNewEntries = async () => {
const newEntries = await this.getEntriesNewerThan(lastTimestamp);
if (newEntries.length > 0) {
for (const entry of newEntries) {
observer.next(entry);
}
lastTimestamp = new Date(newEntries[newEntries.length - 1].timestamp).getTime();
}
};
const startPolling = async () => {
await fetchLastEntryTimestamp();
const intervalId = setInterval(fetchNewEntries, pollInterval);
// Cleanup on unsubscribe
return () => clearInterval(intervalId);
};
startPolling().catch((err) => observer.error(err));
});
}
2022-03-07 14:49:47 +00:00
}