|
@@ -48,8 +48,6 @@ WORKSPACE_DEFAULT = os.path.join(
|
|
|
|
|
|
|
|
# 探查硬编码:增量同步标准锚点字段(推行后端命名标准)
|
|
# 探查硬编码:增量同步标准锚点字段(推行后端命名标准)
|
|
|
ANCHOR_FIELDS = ('create_time', 'update_time')
|
|
ANCHOR_FIELDS = ('create_time', 'update_time')
|
|
|
-# 近期窗口:按 PK 倒序取 N 行验证 update_time 是否近期连续维护
|
|
|
|
|
-PROBE_RECENT_LIMIT = 100
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def resolve_datasource(ds_ref):
|
|
def resolve_datasource(ds_ref):
|
|
@@ -104,13 +102,14 @@ def query_columns_full(conn, schema, table):
|
|
|
|
|
|
|
|
|
|
|
|
|
def probe_table(conn, schema, table, full_rows):
|
|
def probe_table(conn, schema, table, full_rows):
|
|
|
- """对表做行数 + PK + 锚点(主键序范围 / 最早非空 update_time / 近期连续)+ 软删命中。
|
|
|
|
|
|
|
+ """对表做行数 + PK + 锚点(主键序范围 / 最早非空 update_time)+ 软删命中。
|
|
|
|
|
|
|
|
- 行数:pg_class.reltuples 估值
|
|
- 行数:pg_class.reltuples 估值
|
|
|
- PK:单/复合/无 + 是否自增(attidentity + default 表达式 nextval 双判)
|
|
- PK:单/复合/无 + 是否自增(attidentity + default 表达式 nextval 双判)
|
|
|
- create_time 主键序范围:单列自增 PK + create_time 存在;ORDER BY pk ASC/DESC LIMIT 1
|
|
- create_time 主键序范围:单列自增 PK + create_time 存在;ORDER BY pk ASC/DESC LIMIT 1
|
|
|
- update_time 最早非空:全表 min(update_time),加 statement_timeout 60s 保护
|
|
- update_time 最早非空:全表 min(update_time),加 statement_timeout 60s 保护
|
|
|
- - update_time 近期连续:单 PK 时 ORDER BY pk DESC LIMIT 100 验证近期是否连续维护
|
|
|
|
|
|
|
+ (非空即视为业务方已启用 update_time 维护;近期 100 行非空率不能区分
|
|
|
|
|
+ "业务方未维护"和"近期创建的行还没被 update 过",所以不再探查)
|
|
|
- 软删:full_rows 筛 'del' 子串(不区分大小写)
|
|
- 软删:full_rows 筛 'del' 子串(不区分大小写)
|
|
|
"""
|
|
"""
|
|
|
cur = conn.cursor()
|
|
cur = conn.cursor()
|
|
@@ -181,17 +180,6 @@ def probe_table(conn, schema, table, full_rows):
|
|
|
except Exception:
|
|
except Exception:
|
|
|
pass
|
|
pass
|
|
|
|
|
|
|
|
- recent_total = None
|
|
|
|
|
- recent_update_notnull = None
|
|
|
|
|
- if len(pk_cols) == 1 and update_exists:
|
|
|
|
|
- cur.execute(
|
|
|
|
|
- 'SELECT count(*), count("update_time") FROM '
|
|
|
|
|
- '(SELECT update_time FROM "{s}"."{t}" ORDER BY "{p}" DESC LIMIT {n}) AS sub'
|
|
|
|
|
- .format(s=schema, t=table, p=pk_cols[0], n=PROBE_RECENT_LIMIT))
|
|
|
|
|
- r = cur.fetchone()
|
|
|
|
|
- recent_total = int(r[0])
|
|
|
|
|
- recent_update_notnull = int(r[1])
|
|
|
|
|
-
|
|
|
|
|
del_candidates = sorted(r[1] for r in full_rows if 'del' in r[1].lower())
|
|
del_candidates = sorted(r[1] for r in full_rows if 'del' in r[1].lower())
|
|
|
|
|
|
|
|
return {
|
|
return {
|
|
@@ -204,8 +192,6 @@ def probe_table(conn, schema, table, full_rows):
|
|
|
'update_exists': update_exists,
|
|
'update_exists': update_exists,
|
|
|
'update_earliest': update_earliest,
|
|
'update_earliest': update_earliest,
|
|
|
'update_earliest_timeout': update_earliest_timeout,
|
|
'update_earliest_timeout': update_earliest_timeout,
|
|
|
- 'recent_total': recent_total,
|
|
|
|
|
- 'recent_update_notnull': recent_update_notnull,
|
|
|
|
|
'del_candidates': del_candidates,
|
|
'del_candidates': del_candidates,
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -239,18 +225,13 @@ def render_probe_md(stats):
|
|
|
if not stats['update_exists']:
|
|
if not stats['update_exists']:
|
|
|
lines.append(' - `update_time`:缺失')
|
|
lines.append(' - `update_time`:缺失')
|
|
|
else:
|
|
else:
|
|
|
- parts = ['`update_time`:存在']
|
|
|
|
|
if stats['update_earliest_timeout']:
|
|
if stats['update_earliest_timeout']:
|
|
|
- parts.append('最早非空(查询超时)')
|
|
|
|
|
|
|
+ tail = '最早非空(查询超时)'
|
|
|
elif stats['update_earliest'] is not None:
|
|
elif stats['update_earliest'] is not None:
|
|
|
- parts.append('最早非空 {}'.format(stats['update_earliest']))
|
|
|
|
|
|
|
+ tail = '最早非空 {}'.format(stats['update_earliest'])
|
|
|
else:
|
|
else:
|
|
|
- parts.append('最早非空(全 NULL)')
|
|
|
|
|
- if stats['recent_total'] is not None and stats['recent_total'] > 0:
|
|
|
|
|
- rpct = 100.0 * stats['recent_update_notnull'] / stats['recent_total']
|
|
|
|
|
- parts.append('近期 {} 行非空率 {:.1f}%'.format(
|
|
|
|
|
- stats['recent_total'], rpct))
|
|
|
|
|
- lines.append(' - ' + ';'.join(parts))
|
|
|
|
|
|
|
+ tail = '最早非空(全 NULL,业务方未启用维护)'
|
|
|
|
|
+ lines.append(' - `update_time`:存在;' + tail)
|
|
|
|
|
|
|
|
if stats['del_candidates']:
|
|
if stats['del_candidates']:
|
|
|
lines.append('- 软删字段(含 `del` 子串):' + ', '.join(
|
|
lines.append('- 软删字段(含 `del` 子串):' + ', '.join(
|