mirror of
https://github.com/Syngnat/GoNavi.git
synced 2026-08-15 19:24:23 +08:00
fix(sync): constrain time-series targets to append writes
This commit is contained in:
@@ -19,6 +19,7 @@ const supportedCapability: DataSyncRouteCapability = {
|
||||
supportsAutoCreate: true,
|
||||
supportsAutoAddColumns: true,
|
||||
requiresExistingTarget: false,
|
||||
supportsMutations: true,
|
||||
supportsCdc: true,
|
||||
};
|
||||
|
||||
@@ -140,6 +141,43 @@ describe('DataSyncTaskEditor delivery stage', () => {
|
||||
});
|
||||
});
|
||||
|
||||
it('constrains append-only targets to inserts and removes delete propagation', async () => {
|
||||
const mapping = {
|
||||
...createDataSyncTableMapping('timeseries:mapping:1', 'orders', 'orders'),
|
||||
keyColumns: ['id'],
|
||||
};
|
||||
const base = createDataSyncTaskDraft({ id: 'timeseries', kind: 'reconcile' });
|
||||
const task = reviseDataSyncTask(base, {
|
||||
source: endpoint('source'),
|
||||
target: endpoint('target'),
|
||||
mappings: [mapping],
|
||||
delivery: {
|
||||
...base.delivery,
|
||||
writeMode: 'upsert',
|
||||
retryLimit: 3,
|
||||
propagateDeletes: true,
|
||||
},
|
||||
});
|
||||
const { renderer, onPatch } = await renderDelivery(task, vi.fn(), {
|
||||
...supportedCapability,
|
||||
supportsMutations: false,
|
||||
});
|
||||
|
||||
expect(JSON.stringify(renderer.toJSON())).toContain('当前时序目标仅支持追加写入');
|
||||
expect(renderer.root.findAllByProps({ 'data-delete-propagation': 'true' })).toHaveLength(0);
|
||||
const upsert = renderer.root
|
||||
.findAllByType('option')
|
||||
.find((option) => option.props.value === 'upsert')!;
|
||||
expect(upsert.props.disabled).toBe(true);
|
||||
expect(onPatch).toHaveBeenCalledWith({
|
||||
delivery: expect.objectContaining({
|
||||
writeMode: 'append',
|
||||
retryLimit: 0,
|
||||
propagateDeletes: false,
|
||||
}),
|
||||
});
|
||||
});
|
||||
|
||||
it('reveals migration schema controls only for supported implicit same-name mappings', async () => {
|
||||
const implicitMapping = {
|
||||
...createDataSyncTableMapping('migration:mapping:1', 'public.orders', 'public.orders'),
|
||||
|
||||
@@ -300,6 +300,8 @@ const DeliveryStage: React.FC<{
|
||||
(task.kind !== 'cdc' || capability.supportsCdc === true));
|
||||
const rowIsolationAvailable =
|
||||
routeCanWrite && canUseDataSyncRowErrorIsolation(task);
|
||||
const appendOnlyTarget =
|
||||
capability.level !== 'unknown' && capability.supportsMutations === false;
|
||||
const enabledMappings = task.mappings.filter((mapping) => mapping.enabled);
|
||||
const allEnabledMappingsHaveKeys =
|
||||
enabledMappings.length > 0 &&
|
||||
@@ -308,6 +310,7 @@ const DeliveryStage: React.FC<{
|
||||
);
|
||||
const canPropagateDeletes =
|
||||
routeCanWrite &&
|
||||
!appendOnlyTarget &&
|
||||
task.delivery.writeMode === 'upsert' &&
|
||||
((task.kind === 'reconcile' &&
|
||||
task.incremental.mode === 'snapshot' &&
|
||||
@@ -370,6 +373,10 @@ const DeliveryStage: React.FC<{
|
||||
if (!canPropagateDeletes && task.delivery.propagateDeletes) {
|
||||
patch.propagateDeletes = false;
|
||||
}
|
||||
if (appendOnlyTarget && task.delivery.writeMode !== 'append') {
|
||||
patch.writeMode = 'append';
|
||||
patch.retryLimit = 0;
|
||||
}
|
||||
if (
|
||||
structureCapabilityResolved &&
|
||||
!canAutoAddColumns &&
|
||||
@@ -391,6 +398,7 @@ const DeliveryStage: React.FC<{
|
||||
canAutoAddColumns,
|
||||
canCreateIndexes,
|
||||
canPropagateDeletes,
|
||||
appendOnlyTarget,
|
||||
onPatch,
|
||||
readOnly,
|
||||
rowIsolationAvailable,
|
||||
@@ -400,6 +408,7 @@ const DeliveryStage: React.FC<{
|
||||
task.delivery.createIndexes,
|
||||
task.delivery.errorPolicy,
|
||||
task.delivery.propagateDeletes,
|
||||
task.delivery.writeMode,
|
||||
]);
|
||||
|
||||
if (readOnly) {
|
||||
@@ -428,6 +437,11 @@ const DeliveryStage: React.FC<{
|
||||
</div>
|
||||
</header>
|
||||
<div className="gn-data-sync-delivery-main">
|
||||
{appendOnlyTarget ? (
|
||||
<p className="gn-data-sync-inline-note" role="note" data-append-only-target="true">
|
||||
{t('delivery.append_only_target_note')}
|
||||
</p>
|
||||
) : null}
|
||||
<Field label={t('delivery.write_mode')}>
|
||||
<select
|
||||
className="gn-data-sync-control"
|
||||
@@ -451,7 +465,11 @@ const DeliveryStage: React.FC<{
|
||||
>
|
||||
{t('delivery.write.append')}
|
||||
</option>
|
||||
<option value="upsert">{t('delivery.write.upsert')}</option>
|
||||
<option value="upsert" disabled={appendOnlyTarget}>
|
||||
{appendOnlyTarget
|
||||
? t('delivery.write.upsert_unavailable')
|
||||
: t('delivery.write.upsert')}
|
||||
</option>
|
||||
<option value="overwrite" disabled>
|
||||
{t('delivery.write.overwrite_unavailable')}
|
||||
</option>
|
||||
|
||||
@@ -46,6 +46,7 @@ const EMPTY_CAPABILITY: DataSyncRouteCapability = {
|
||||
level: 'unknown',
|
||||
canExecute: false,
|
||||
supportsAutoCreate: false,
|
||||
supportsMutations: false,
|
||||
supportsCdc: false,
|
||||
};
|
||||
|
||||
|
||||
@@ -178,6 +178,7 @@ const unresolvedCapability: DataSyncRouteCapability = {
|
||||
level: 'unknown',
|
||||
canExecute: false,
|
||||
supportsAutoCreate: false,
|
||||
supportsMutations: false,
|
||||
supportsCdc: false,
|
||||
};
|
||||
|
||||
|
||||
@@ -275,6 +275,7 @@ export type DataSyncRouteCapability = {
|
||||
supportsAutoCreate: boolean;
|
||||
supportsAutoAddColumns?: boolean;
|
||||
requiresExistingTarget?: boolean;
|
||||
supportsMutations?: boolean;
|
||||
supportsCdc: boolean;
|
||||
};
|
||||
|
||||
|
||||
@@ -162,6 +162,7 @@ const zhCN = {
|
||||
'delivery.write.append': '追加',
|
||||
'delivery.write.append_desc': '每次运行都新增记录;重复运行可能产生重复数据,系统会自动关闭重试。',
|
||||
'delivery.write.upsert': '新增或更新',
|
||||
'delivery.write.upsert_unavailable': '新增或更新(当前目标不支持)',
|
||||
'delivery.write.upsert_desc': '按主键或唯一键识别同一行:不存在则新增,已存在则更新。',
|
||||
'delivery.write.overwrite': '覆盖',
|
||||
'delivery.write.overwrite_unavailable': '覆盖(当前任务系统暂不支持)',
|
||||
@@ -182,6 +183,7 @@ const zhCN = {
|
||||
'delivery.retry_backoff': '重试退避(毫秒)',
|
||||
'delivery.row_isolation_note': '当前配置必须遇错停止。逐行跳过只适用于写入已有关系表,且不能同时自动建表、修改结构、同步删除或覆盖写入。',
|
||||
'delivery.append_retry_note': '追加写入可能已部分落库,为避免重复写入,自动重试固定为 0。',
|
||||
'delivery.append_only_target_note': '当前时序目标仅支持追加写入(INSERT);更新和删除不会同步。',
|
||||
'delivery.propagate_deletes': '传播源端删除',
|
||||
'delivery.delete_policy_title': '目标端删除策略',
|
||||
'delivery.delete_risk_badge': '高风险',
|
||||
@@ -576,6 +578,7 @@ const enUS: Record<DataSyncWorkbenchTextKey, string> = {
|
||||
'delivery.write.append': 'Append',
|
||||
'delivery.write.append_desc': 'Each run inserts new rows. Re-running may create duplicates, so automatic retries are disabled.',
|
||||
'delivery.write.upsert': 'Insert or update',
|
||||
'delivery.write.upsert_unavailable': 'Insert or update (not supported by this target)',
|
||||
'delivery.write.upsert_desc': 'Use a primary or unique key to identify a row: insert when missing and update when present.',
|
||||
'delivery.write.overwrite': 'Overwrite',
|
||||
'delivery.write.overwrite_unavailable': 'Overwrite (not yet supported for jobs)',
|
||||
@@ -596,6 +599,7 @@ const enUS: Record<DataSyncWorkbenchTextKey, string> = {
|
||||
'delivery.retry_backoff': 'Retry backoff (ms)',
|
||||
'delivery.row_isolation_note': 'This configuration must stop on error. Row-level skip is available only for existing relational targets without auto-create, schema changes, delete propagation, or overwrite.',
|
||||
'delivery.append_retry_note': 'Append writes may be partially committed. Automatic retries stay at zero to prevent duplicates.',
|
||||
'delivery.append_only_target_note': 'This time-series target supports append/INSERT-only delivery; updates and deletes are not synchronized.',
|
||||
'delivery.propagate_deletes': 'Propagate source deletes',
|
||||
'delivery.delete_policy_title': 'Target deletion policy',
|
||||
'delivery.delete_risk_badge': 'High risk',
|
||||
|
||||
@@ -103,6 +103,7 @@ describe('data sync Wails DTO boundary', () => {
|
||||
supportsAutoCreate: true,
|
||||
supportsAutoAddColumns: true,
|
||||
requiresExistingTarget: true,
|
||||
supportsMutations: true,
|
||||
}),
|
||||
).toMatchObject({
|
||||
level: 'full',
|
||||
@@ -110,6 +111,7 @@ describe('data sync Wails DTO boundary', () => {
|
||||
supportsAutoCreate: true,
|
||||
supportsAutoAddColumns: true,
|
||||
requiresExistingTarget: true,
|
||||
supportsMutations: true,
|
||||
});
|
||||
expect(
|
||||
decodeRouteCapability({
|
||||
@@ -120,6 +122,7 @@ describe('data sync Wails DTO boundary', () => {
|
||||
).toMatchObject({
|
||||
supportsAutoAddColumns: false,
|
||||
requiresExistingTarget: false,
|
||||
supportsMutations: false,
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -831,6 +831,10 @@ export const decodeRouteCapability = (value: unknown): DataSyncRouteCapability =
|
||||
capability.requiresExistingTarget,
|
||||
'DataSyncCapabilityResolve.data.requiresExistingTarget',
|
||||
),
|
||||
supportsMutations: optionalBoolean(
|
||||
capability.supportsMutations,
|
||||
'DataSyncCapabilityResolve.data.supportsMutations',
|
||||
),
|
||||
supportsCdc: false,
|
||||
};
|
||||
};
|
||||
@@ -899,6 +903,7 @@ export const decodeDataSyncPreflight = (
|
||||
supportsAutoCreate: false,
|
||||
supportsAutoAddColumns: false,
|
||||
requiresExistingTarget: false,
|
||||
supportsMutations: false,
|
||||
supportsCdc: false,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -140,6 +140,7 @@ const UNKNOWN_CAPABILITY: DataSyncRouteCapability = {
|
||||
level: 'unknown',
|
||||
canExecute: false,
|
||||
supportsAutoCreate: false,
|
||||
supportsMutations: false,
|
||||
supportsCdc: false,
|
||||
};
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"GoNavi-Wails/internal/sync"
|
||||
"GoNavi-Wails/internal/syncjob"
|
||||
)
|
||||
|
||||
@@ -162,3 +163,21 @@ func TestDataSyncJobPreflightDiscardsCallerSuppliedApproval(t *testing.T) {
|
||||
t.Fatal("preflight must not trust or echo a caller-supplied approval")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAppendOnlyTargetPreflightIssuesBlockMutations(t *testing.T) {
|
||||
definition := approvalTestDefinition()
|
||||
definition.Options.SyncMode = "insert_update"
|
||||
definition.Options.PropagateDeletes = true
|
||||
issues := appendOnlyTargetPreflightIssues(definition, sync.MigrationCapability{
|
||||
TargetType: "tdengine",
|
||||
SupportsMutations: false,
|
||||
})
|
||||
if len(issues) != 2 || issues[0].Code != "append_only_target_requires_insert_only" || issues[1].Code != "append_only_target_delete_unsupported" {
|
||||
t.Fatalf("unexpected append-only target issues: %#v", issues)
|
||||
}
|
||||
|
||||
definition.Kind = syncjob.JobKindCompare
|
||||
if issues := appendOnlyTargetPreflightIssues(definition, sync.MigrationCapability{TargetType: "tdengine"}); len(issues) != 0 {
|
||||
t.Fatalf("compare task must not be blocked by write capability: %#v", issues)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -58,6 +58,7 @@ func (a *App) preflightDataSyncJob(input syncjob.JobDefinition, now time.Time) D
|
||||
message := fmt.Sprintf("migration route %s -> %s is %s", result.Capability.SourceType, result.Capability.TargetType, result.Capability.SupportLevel)
|
||||
result.Issues = append(result.Issues, preflightIssue("route_unsupported", DataSyncJobPreflightBlocker, "endpoints", message, ""))
|
||||
}
|
||||
result.Issues = append(result.Issues, appendOnlyTargetPreflightIssues(definition, result.Capability)...)
|
||||
if definition.Kind != syncjob.JobKindCompare {
|
||||
for _, mapping := range definition.Mappings {
|
||||
if !mapping.Enabled {
|
||||
@@ -225,6 +226,32 @@ func (a *App) preflightDataSyncJob(input syncjob.JobDefinition, now time.Time) D
|
||||
return finishDataSyncJobPreflight(result)
|
||||
}
|
||||
|
||||
func appendOnlyTargetPreflightIssues(definition syncjob.JobDefinition, capability sync.MigrationCapability) []DataSyncJobPreflightIssue {
|
||||
if definition.Kind == syncjob.JobKindCompare || capability.SupportsMutations {
|
||||
return nil
|
||||
}
|
||||
issues := make([]DataSyncJobPreflightIssue, 0, 2)
|
||||
if !strings.EqualFold(definition.Options.SyncMode, "insert_only") {
|
||||
issues = append(issues, preflightIssue(
|
||||
"append_only_target_requires_insert_only",
|
||||
DataSyncJobPreflightBlocker,
|
||||
"delivery",
|
||||
fmt.Sprintf("%s targets support append/INSERT-only delivery; updates are not supported", capability.TargetType),
|
||||
"",
|
||||
))
|
||||
}
|
||||
if definition.Options.PropagateDeletes {
|
||||
issues = append(issues, preflightIssue(
|
||||
"append_only_target_delete_unsupported",
|
||||
DataSyncJobPreflightBlocker,
|
||||
"delivery",
|
||||
fmt.Sprintf("%s targets support append/INSERT-only delivery; deletes are not supported", capability.TargetType),
|
||||
"",
|
||||
))
|
||||
}
|
||||
return issues
|
||||
}
|
||||
|
||||
func (a *App) preflightDataSyncMappings(definition syncjob.JobDefinition, source, target resolvedDataSyncJobEndpoint) []DataSyncJobPreflightIssue {
|
||||
issues := make([]DataSyncJobPreflightIssue, 0)
|
||||
for _, mapping := range definition.Mappings {
|
||||
|
||||
@@ -16,6 +16,7 @@ type MigrationCapability struct {
|
||||
SupportsAutoCreate bool `json:"supportsAutoCreate"`
|
||||
SupportsAutoAddColumns bool `json:"supportsAutoAddColumns"`
|
||||
RequiresExistingTarget bool `json:"requiresExistingTarget"`
|
||||
SupportsMutations bool `json:"supportsMutations"`
|
||||
}
|
||||
|
||||
// ResolveMigrationCapability returns the migration runtime's effective support
|
||||
@@ -28,6 +29,12 @@ func ResolveMigrationCapability(sourceConfig connection.ConnectionConfig, target
|
||||
TargetType: targetType,
|
||||
SourceModel: classifyMigrationDataModel(sourceType),
|
||||
TargetModel: classifyMigrationDataModel(targetType),
|
||||
// Most writable targets support the job engine's update/delete semantics.
|
||||
// Time-series targets are deliberately narrowed below.
|
||||
SupportsMutations: true,
|
||||
}
|
||||
if targetType == "tdengine" || targetType == "iotdb" {
|
||||
capability.SupportsMutations = false
|
||||
}
|
||||
syncConfig := SyncConfig{
|
||||
SourceConfig: sourceConfig,
|
||||
|
||||
@@ -21,6 +21,7 @@ func TestResolveMigrationCapability_MySQLToPostgresUsesFullPlanner(t *testing.T)
|
||||
SupportsAutoCreate: true,
|
||||
SupportsAutoAddColumns: true,
|
||||
RequiresExistingTarget: false,
|
||||
SupportsMutations: true,
|
||||
}
|
||||
|
||||
if got != want {
|
||||
@@ -44,6 +45,7 @@ func TestResolveMigrationCapability_PostgresToKingbaseUsesSameFamilyPlanner(t *t
|
||||
SupportsAutoCreate: true,
|
||||
SupportsAutoAddColumns: true,
|
||||
RequiresExistingTarget: false,
|
||||
SupportsMutations: true,
|
||||
}
|
||||
|
||||
if got != want {
|
||||
@@ -67,6 +69,7 @@ func TestResolveMigrationCapability_OracleToSQLServerUsesExistingTargetCompatibi
|
||||
SupportsAutoCreate: false,
|
||||
SupportsAutoAddColumns: false,
|
||||
RequiresExistingTarget: true,
|
||||
SupportsMutations: true,
|
||||
}
|
||||
|
||||
if got != want {
|
||||
@@ -90,6 +93,7 @@ func TestResolveMigrationCapability_MongoToOracleReportsPlannedNonExecutablePath
|
||||
SupportsAutoCreate: false,
|
||||
SupportsAutoAddColumns: false,
|
||||
RequiresExistingTarget: true,
|
||||
SupportsMutations: true,
|
||||
}
|
||||
|
||||
if got != want {
|
||||
@@ -113,6 +117,7 @@ func TestResolveMigrationCapability_RedisToMongoReportsFullKeyspaceBridge(t *tes
|
||||
SupportsAutoCreate: true,
|
||||
SupportsAutoAddColumns: false,
|
||||
RequiresExistingTarget: false,
|
||||
SupportsMutations: true,
|
||||
}
|
||||
|
||||
if got != want {
|
||||
@@ -136,6 +141,7 @@ func TestResolveMigrationCapability_MongoToRedisReportsFullKeyspaceBridge(t *tes
|
||||
SupportsAutoCreate: true,
|
||||
SupportsAutoAddColumns: false,
|
||||
RequiresExistingTarget: false,
|
||||
SupportsMutations: true,
|
||||
}
|
||||
|
||||
if got != want {
|
||||
@@ -159,6 +165,7 @@ func TestResolveMigrationCapability_CustomPostgresUsesResolvedDriverFamily(t *te
|
||||
SupportsAutoCreate: true,
|
||||
SupportsAutoAddColumns: true,
|
||||
RequiresExistingTarget: false,
|
||||
SupportsMutations: true,
|
||||
}
|
||||
|
||||
if got != want {
|
||||
@@ -182,6 +189,7 @@ func TestResolveMigrationCapability_KafkaToQdrantIsUnsupported(t *testing.T) {
|
||||
SupportsAutoCreate: false,
|
||||
SupportsAutoAddColumns: false,
|
||||
RequiresExistingTarget: true,
|
||||
SupportsMutations: true,
|
||||
}
|
||||
|
||||
if got != want {
|
||||
@@ -200,6 +208,20 @@ func TestResolveMigrationCapability_RedisToNonMongoTargetIsUnsupported(t *testin
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveMigrationCapability_TimeSeriesTargetsAreAppendOnly(t *testing.T) {
|
||||
for _, targetType := range []string{"tdengine", "iotdb"} {
|
||||
t.Run(targetType, func(t *testing.T) {
|
||||
capability := ResolveMigrationCapability(
|
||||
connection.ConnectionConfig{Type: "mysql"},
|
||||
connection.ConnectionConfig{Type: targetType},
|
||||
)
|
||||
if !capability.CanExecute || capability.SupportsMutations {
|
||||
t.Fatalf("expected %s target to be executable and append-only, got %+v", targetType, capability)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveMigrationCapability_GoldenDBUsesMySQLPlannerFamily(t *testing.T) {
|
||||
got := ResolveMigrationCapability(
|
||||
connection.ConnectionConfig{Type: "goldendb"},
|
||||
|
||||
Reference in New Issue
Block a user