feat: Split HLC storage into separate timestamp, counter, and node ID columns (#28)
Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: marcelomendoncasoares <6807059+marcelomendoncasoares@users.noreply.github.com>
This commit is contained in:
parent
0c1d63eec7
commit
027aec5ed0
3 changed files with 117 additions and 18 deletions
|
|
@ -80,8 +80,14 @@ class CrdtNormalizedDataTable extends Table {
|
||||||
/// Unique identifier for the row in the table.
|
/// Unique identifier for the row in the table.
|
||||||
late final rowId = text()();
|
late final rowId = text()();
|
||||||
|
|
||||||
/// Hybrid Logical Clock timestamp of the data update.
|
/// Hybrid Logical Clock timestamp component (milliseconds since epoch).
|
||||||
late final hlcTimestamp = text().map(hlcConverter)();
|
late final hlcTimestamp = integer()();
|
||||||
|
|
||||||
|
/// Hybrid Logical Clock counter component.
|
||||||
|
late final hlcCounter = integer()();
|
||||||
|
|
||||||
|
/// Hybrid Logical Clock node ID component.
|
||||||
|
late final hlcNodeId = text()();
|
||||||
|
|
||||||
/// Raw value of the data.
|
/// Raw value of the data.
|
||||||
late final rawValue = sqliteAny().nullable()();
|
late final rawValue = sqliteAny().nullable()();
|
||||||
|
|
|
||||||
|
|
@ -146,10 +146,66 @@ ${columnInserts.join('\n UNION ALL\n')}
|
||||||
/// same format as the `hlcConverter.toSql` method.
|
/// same format as the `hlcConverter.toSql` method.
|
||||||
String _getHlcJsonExprSql() {
|
String _getHlcJsonExprSql() {
|
||||||
return "printf('%015d', \"last_timestamp\") || '-' || "
|
return "printf('%015d', \"last_timestamp\") || '-' || "
|
||||||
"printf('%05x', \"counter\") || '-' || "
|
"printf('%04X', \"counter\") || '-' || "
|
||||||
"'${crdtDb.nodeId}'";
|
"'${crdtDb.nodeId}'";
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Returns SQL fragment that combines the split HLC columns (hlc_timestamp,
|
||||||
|
/// hlc_counter, hlc_node_id) back into a single text field.
|
||||||
|
String _getHlcCombineExprSql() {
|
||||||
|
return "printf('%015d', \"hlc_timestamp\") || '-' || "
|
||||||
|
"printf('%04X', \"hlc_counter\") || '-' || "
|
||||||
|
'"hlc_node_id"';
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns SQL expression that converts a 4-character hex string to integer.
|
||||||
|
/// Used to parse the counter from the HLC text format.
|
||||||
|
String _hexToIntExpr(String hexString) {
|
||||||
|
// Convert each hex digit (0-9, A-F) to its decimal value and combine
|
||||||
|
return '''(
|
||||||
|
CASE SUBSTR($hexString, 1, 1)
|
||||||
|
WHEN '0' THEN 0 WHEN '1' THEN 1 WHEN '2' THEN 2 WHEN '3' THEN 3
|
||||||
|
WHEN '4' THEN 4 WHEN '5' THEN 5 WHEN '6' THEN 6 WHEN '7' THEN 7
|
||||||
|
WHEN '8' THEN 8 WHEN '9' THEN 9
|
||||||
|
WHEN 'A' THEN 10 WHEN 'B' THEN 11 WHEN 'C' THEN 12 WHEN 'D' THEN 13
|
||||||
|
WHEN 'E' THEN 14 WHEN 'F' THEN 15
|
||||||
|
WHEN 'a' THEN 10 WHEN 'b' THEN 11 WHEN 'c' THEN 12 WHEN 'd' THEN 13
|
||||||
|
WHEN 'e' THEN 14 WHEN 'f' THEN 15
|
||||||
|
ELSE 0
|
||||||
|
END * 4096 +
|
||||||
|
CASE SUBSTR($hexString, 2, 1)
|
||||||
|
WHEN '0' THEN 0 WHEN '1' THEN 1 WHEN '2' THEN 2 WHEN '3' THEN 3
|
||||||
|
WHEN '4' THEN 4 WHEN '5' THEN 5 WHEN '6' THEN 6 WHEN '7' THEN 7
|
||||||
|
WHEN '8' THEN 8 WHEN '9' THEN 9
|
||||||
|
WHEN 'A' THEN 10 WHEN 'B' THEN 11 WHEN 'C' THEN 12 WHEN 'D' THEN 13
|
||||||
|
WHEN 'E' THEN 14 WHEN 'F' THEN 15
|
||||||
|
WHEN 'a' THEN 10 WHEN 'b' THEN 11 WHEN 'c' THEN 12 WHEN 'd' THEN 13
|
||||||
|
WHEN 'e' THEN 14 WHEN 'f' THEN 15
|
||||||
|
ELSE 0
|
||||||
|
END * 256 +
|
||||||
|
CASE SUBSTR($hexString, 3, 1)
|
||||||
|
WHEN '0' THEN 0 WHEN '1' THEN 1 WHEN '2' THEN 2 WHEN '3' THEN 3
|
||||||
|
WHEN '4' THEN 4 WHEN '5' THEN 5 WHEN '6' THEN 6 WHEN '7' THEN 7
|
||||||
|
WHEN '8' THEN 8 WHEN '9' THEN 9
|
||||||
|
WHEN 'A' THEN 10 WHEN 'B' THEN 11 WHEN 'C' THEN 12 WHEN 'D' THEN 13
|
||||||
|
WHEN 'E' THEN 14 WHEN 'F' THEN 15
|
||||||
|
WHEN 'a' THEN 10 WHEN 'b' THEN 11 WHEN 'c' THEN 12 WHEN 'd' THEN 13
|
||||||
|
WHEN 'e' THEN 14 WHEN 'f' THEN 15
|
||||||
|
ELSE 0
|
||||||
|
END * 16 +
|
||||||
|
CASE SUBSTR($hexString, 4, 1)
|
||||||
|
WHEN '0' THEN 0 WHEN '1' THEN 1 WHEN '2' THEN 2 WHEN '3' THEN 3
|
||||||
|
WHEN '4' THEN 4 WHEN '5' THEN 5 WHEN '6' THEN 6 WHEN '7' THEN 7
|
||||||
|
WHEN '8' THEN 8 WHEN '9' THEN 9
|
||||||
|
WHEN 'A' THEN 10 WHEN 'B' THEN 11 WHEN 'C' THEN 12 WHEN 'D' THEN 13
|
||||||
|
WHEN 'E' THEN 14 WHEN 'F' THEN 15
|
||||||
|
WHEN 'a' THEN 10 WHEN 'b' THEN 11 WHEN 'c' THEN 12 WHEN 'd' THEN 13
|
||||||
|
WHEN 'e' THEN 14 WHEN 'f' THEN 15
|
||||||
|
ELSE 0
|
||||||
|
END
|
||||||
|
)''';
|
||||||
|
}
|
||||||
|
|
||||||
/// Returns SQL statements to create the CRDT data view and its INSTEAD OF
|
/// Returns SQL statements to create the CRDT data view and its INSTEAD OF
|
||||||
/// triggers, and the HLC canonical view with its INSTEAD OF trigger.
|
/// triggers, and the HLC canonical view with its INSTEAD OF trigger.
|
||||||
///
|
///
|
||||||
|
|
@ -180,38 +236,72 @@ ${columnInserts.join('\n UNION ALL\n')}
|
||||||
|
|
||||||
String get _normalizedTableName => crdtDb.crdtNormalizedDataTable.actualTableName;
|
String get _normalizedTableName => crdtDb.crdtNormalizedDataTable.actualTableName;
|
||||||
|
|
||||||
/// CREATE VIEW for the CRDT data view: selects all columns from the
|
/// CREATE VIEW for the CRDT data view: combines the split HLC columns from
|
||||||
/// normalized table so reads against the view hit the real table.
|
/// the normalized table back into a single hlc_timestamp text field.
|
||||||
String _createCrdtDataViewStatement() =>
|
String _createCrdtDataViewStatement() =>
|
||||||
'''
|
'''
|
||||||
CREATE VIEW IF NOT EXISTS "$crdtDataTableName" AS
|
CREATE VIEW IF NOT EXISTS "$crdtDataTableName" AS
|
||||||
SELECT ${crdtDataAllColumns.join(', ')} FROM "$_normalizedTableName";''';
|
SELECT
|
||||||
|
"user_id",
|
||||||
|
"table_name",
|
||||||
|
"column_name",
|
||||||
|
"row_id",
|
||||||
|
${_getHlcCombineExprSql()} AS "hlc_timestamp",
|
||||||
|
"raw_value"
|
||||||
|
FROM "$_normalizedTableName";''';
|
||||||
|
|
||||||
/// INSTEAD OF INSERT on the CRDT data view: insert into the normalized table,
|
/// INSTEAD OF INSERT on the CRDT data view: insert into the normalized table,
|
||||||
/// and on conflict update only when the new row has a greater hlc_timestamp.
|
/// splitting the hlc_timestamp text into 3 columns, and on conflict update
|
||||||
|
/// only when the new row has a greater hlc_timestamp.
|
||||||
String _createCrdtDataInsteadOfInsertTriggerStatement() =>
|
String _createCrdtDataInsteadOfInsertTriggerStatement() =>
|
||||||
'''
|
'''
|
||||||
CREATE TRIGGER IF NOT EXISTS "${crdtDataTableName}__instead_of_insert"
|
CREATE TRIGGER IF NOT EXISTS "${crdtDataTableName}__instead_of_insert"
|
||||||
INSTEAD OF INSERT ON "$crdtDataTableName"
|
INSTEAD OF INSERT ON "$crdtDataTableName"
|
||||||
BEGIN
|
BEGIN
|
||||||
INSERT INTO "$_normalizedTableName" (${crdtDataAllColumns.join(', ')})
|
INSERT INTO "$_normalizedTableName" (
|
||||||
VALUES (${crdtDataAllColumns.map((c) => 'NEW.$c').join(', ')})
|
"user_id", "table_name", "column_name", "row_id",
|
||||||
ON CONFLICT (${crdtDataPrimaryKeyColumns.join(', ')}) DO UPDATE SET
|
"hlc_timestamp", "hlc_counter", "hlc_node_id", "raw_value"
|
||||||
|
)
|
||||||
|
SELECT
|
||||||
|
NEW."user_id",
|
||||||
|
NEW."table_name",
|
||||||
|
NEW."column_name",
|
||||||
|
NEW."row_id",
|
||||||
|
CAST(SUBSTR(NEW."hlc_timestamp", 1, 15) AS INTEGER),
|
||||||
|
${_hexToIntExpr('SUBSTR(NEW."hlc_timestamp", 17, 4)')},
|
||||||
|
SUBSTR(NEW."hlc_timestamp", 22),
|
||||||
|
NEW."raw_value"
|
||||||
|
ON CONFLICT ("user_id", "table_name", "column_name", "row_id") DO UPDATE SET
|
||||||
"hlc_timestamp" = excluded."hlc_timestamp",
|
"hlc_timestamp" = excluded."hlc_timestamp",
|
||||||
|
"hlc_counter" = excluded."hlc_counter",
|
||||||
|
"hlc_node_id" = excluded."hlc_node_id",
|
||||||
"raw_value" = excluded."raw_value"
|
"raw_value" = excluded."raw_value"
|
||||||
WHERE "hlc_timestamp" < excluded."hlc_timestamp";
|
WHERE "hlc_timestamp" < excluded."hlc_timestamp"
|
||||||
|
OR ("hlc_timestamp" = excluded."hlc_timestamp" AND "hlc_counter" < excluded."hlc_counter");
|
||||||
END;''';
|
END;''';
|
||||||
|
|
||||||
/// INSTEAD OF UPDATE on the CRDT data view: apply the new values to the
|
/// INSTEAD OF UPDATE on the CRDT data view: apply the new values to the
|
||||||
/// matching row in the normalized table (by primary key).
|
/// matching row in the normalized table (by primary key), splitting the
|
||||||
|
/// hlc_timestamp text into 3 columns.
|
||||||
String _createCrdtDataInsteadOfUpdateTriggerStatement() =>
|
String _createCrdtDataInsteadOfUpdateTriggerStatement() =>
|
||||||
'''
|
'''
|
||||||
CREATE TRIGGER IF NOT EXISTS "${crdtDataTableName}__instead_of_update"
|
CREATE TRIGGER IF NOT EXISTS "${crdtDataTableName}__instead_of_update"
|
||||||
INSTEAD OF UPDATE ON "$crdtDataTableName"
|
INSTEAD OF UPDATE ON "$crdtDataTableName"
|
||||||
BEGIN
|
BEGIN
|
||||||
UPDATE "$_normalizedTableName"
|
UPDATE "$_normalizedTableName"
|
||||||
SET ${crdtDataAllColumns.map((c) => '$c = NEW.$c').join(', ')}
|
SET
|
||||||
WHERE ${crdtDataPrimaryKeyColumns.map((c) => '$c = OLD.$c').join(' AND ')};
|
"user_id" = NEW."user_id",
|
||||||
|
"table_name" = NEW."table_name",
|
||||||
|
"column_name" = NEW."column_name",
|
||||||
|
"row_id" = NEW."row_id",
|
||||||
|
"hlc_timestamp" = CAST(SUBSTR(NEW."hlc_timestamp", 1, 15) AS INTEGER),
|
||||||
|
"hlc_counter" = ${_hexToIntExpr('SUBSTR(NEW."hlc_timestamp", 17, 4)')},
|
||||||
|
"hlc_node_id" = SUBSTR(NEW."hlc_timestamp", 22),
|
||||||
|
"raw_value" = NEW."raw_value"
|
||||||
|
WHERE "user_id" = OLD."user_id"
|
||||||
|
AND "table_name" = OLD."table_name"
|
||||||
|
AND "column_name" = OLD."column_name"
|
||||||
|
AND "row_id" = OLD."row_id";
|
||||||
END;''';
|
END;''';
|
||||||
|
|
||||||
/// INSTEAD OF DELETE on the CRDT data view: delete the matching row from the
|
/// INSTEAD OF DELETE on the CRDT data view: delete the matching row from the
|
||||||
|
|
@ -222,7 +312,10 @@ CREATE TRIGGER IF NOT EXISTS "${crdtDataTableName}__instead_of_delete"
|
||||||
INSTEAD OF DELETE ON "$crdtDataTableName"
|
INSTEAD OF DELETE ON "$crdtDataTableName"
|
||||||
BEGIN
|
BEGIN
|
||||||
DELETE FROM "$_normalizedTableName"
|
DELETE FROM "$_normalizedTableName"
|
||||||
WHERE ${crdtDataPrimaryKeyColumns.map((c) => '$c = OLD.$c').join(' AND ')};
|
WHERE "user_id" = OLD."user_id"
|
||||||
|
AND "table_name" = OLD."table_name"
|
||||||
|
AND "column_name" = OLD."column_name"
|
||||||
|
AND "row_id" = OLD."row_id";
|
||||||
END;''';
|
END;''';
|
||||||
|
|
||||||
/// CREATE VIEW for the HLC canonical view: an empty view (WHERE FALSE) used
|
/// CREATE VIEW for the HLC canonical view: an empty view (WHERE FALSE) used
|
||||||
|
|
|
||||||
|
|
@ -57,7 +57,7 @@ BEGIN
|
||||||
"column_name",
|
"column_name",
|
||||||
NEW."id" AS "row_id",
|
NEW."id" AS "row_id",
|
||||||
(
|
(
|
||||||
SELECT (printf('%015d', "last_timestamp") || '-' || printf('%05x', "counter") || '-' || '${migrator.nodeId}')
|
SELECT (printf('%015d', "last_timestamp") || '-' || printf('%04X', "counter") || '-' || '${migrator.nodeId}')
|
||||||
FROM "${migrator.crdtDb.sqlBuilder.hlcStateTableName}"
|
FROM "${migrator.crdtDb.sqlBuilder.hlcStateTableName}"
|
||||||
WHERE "user_id" = '${migrator.userId}'
|
WHERE "user_id" = '${migrator.userId}'
|
||||||
) AS "hlc_timestamp",
|
) AS "hlc_timestamp",
|
||||||
|
|
@ -124,7 +124,7 @@ BEGIN
|
||||||
"column_name",
|
"column_name",
|
||||||
OLD."id" AS "row_id",
|
OLD."id" AS "row_id",
|
||||||
(
|
(
|
||||||
SELECT (printf('%015d', "last_timestamp") || '-' || printf('%05x', "counter") || '-' || '${migrator.nodeId}')
|
SELECT (printf('%015d', "last_timestamp") || '-' || printf('%04X', "counter") || '-' || '${migrator.nodeId}')
|
||||||
FROM "${migrator.crdtDb.sqlBuilder.hlcStateTableName}"
|
FROM "${migrator.crdtDb.sqlBuilder.hlcStateTableName}"
|
||||||
WHERE "user_id" = '${migrator.userId}'
|
WHERE "user_id" = '${migrator.userId}'
|
||||||
) AS "hlc_timestamp",
|
) AS "hlc_timestamp",
|
||||||
|
|
@ -194,7 +194,7 @@ BEGIN
|
||||||
"column_name",
|
"column_name",
|
||||||
OLD."id" AS "row_id",
|
OLD."id" AS "row_id",
|
||||||
(
|
(
|
||||||
SELECT (printf('%015d', "last_timestamp") || '-' || printf('%05x', "counter") || '-' || '${migrator.nodeId}')
|
SELECT (printf('%015d', "last_timestamp") || '-' || printf('%04X', "counter") || '-' || '${migrator.nodeId}')
|
||||||
FROM "${migrator.crdtDb.sqlBuilder.hlcStateTableName}"
|
FROM "${migrator.crdtDb.sqlBuilder.hlcStateTableName}"
|
||||||
WHERE "user_id" = '${migrator.userId}'
|
WHERE "user_id" = '${migrator.userId}'
|
||||||
) AS "hlc_timestamp",
|
) AS "hlc_timestamp",
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue