diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/AnalyticsTableManager.java b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/AnalyticsTableManager.java index de245125919c..25501c12632d 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/AnalyticsTableManager.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/AnalyticsTableManager.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2004-2022, University of Oslo + * Copyright (c) 2004-2026, University of Oslo * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -94,6 +94,29 @@ default boolean hasUpdatedLatestData(Date startDate, Date endDate) { */ void preCreateTables(AnalyticsTableUpdateParams params); + /** + * Indicates whether the given tables are in a state that supports a continuous ("latest" + * partition) update. For engines requiring unique-key analytics tables (e.g. Doris), a main table + * created before that support was introduced remains on the previous key model until a full + * rebuild recreates it, and continuous updates must not be attempted against it. + * + * @param tables the list of {@link AnalyticsTable}. + * @return true if a continuous update can proceed. + */ + default boolean isReadyForContinuousUpdate(List tables) { + return true; + } + + /** + * Checks whether the main table of the given analytics table exists. + * + * @param table the {@link AnalyticsTable}. + * @return true if the main table exists. + */ + default boolean mainTableExists(AnalyticsTable table) { + return true; + } + /** * Removes updated and deleted data from the given tables for "latest" partition update. * diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/AnalyticsTableService.java b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/AnalyticsTableService.java index 3383f6f73ef8..9b5eeb5eccd0 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/AnalyticsTableService.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/AnalyticsTableService.java @@ -49,8 +49,11 @@ public interface AnalyticsTableService { * * @param params the {@link AnalyticsTableUpdateParams}. * @param progress job progress tracking and control flow + * @return true if the analytics tables reflect all source data changed before the start time of + * the update, false if the update was cancelled or aborted, in which case it must not be + * recorded as a successful update of this table type. */ - void create(AnalyticsTableUpdateParams params, JobProgress progress); + boolean create(AnalyticsTableUpdateParams params, JobProgress progress); /** Drops main and staging analytics tables. */ void dropTables(); diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/AbstractJdbcTableManager.java b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/AbstractJdbcTableManager.java index ed300718f0ad..1eb479b6d4c4 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/AbstractJdbcTableManager.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/AbstractJdbcTableManager.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2004-2022, University of Oslo + * Copyright (c) 2004-2026, University of Oslo * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -176,6 +176,11 @@ public void preCreateTables(AnalyticsTableUpdateParams params) {} @Override public void removeUpdatedData(List tables) {} + @Override + public boolean mainTableExists(AnalyticsTable table) { + return tableExists(table.getMainName()); + } + @Override public void createTable(AnalyticsTable table) { createAnalyticsTable(table); @@ -222,33 +227,120 @@ public void createIndex(Index index) { jdbcTemplate.execute(sql); } + /** How staged data for a table update is published to the main table. */ + private enum TableUpdateAction { + REPLACE_TABLE, + REPLACE_PARTITIONS, + MERGE_ROWS + } + @Override public void swapTable(AnalyticsTableUpdateParams params, AnalyticsTable table) { - boolean tableExists = tableExists(table.getMainName()); - boolean skipMasterTable = - params.isPartialUpdate() && tableExists && table.getTableType().isLatestPartition(); + TableUpdateAction action = resolveUpdateAction(params, table); + + log.info("Swapping table: '{}', action: '{}'", table.getMainName(), action); + + switch (action) { + case REPLACE_TABLE -> replaceMainTable(table); + case REPLACE_PARTITIONS -> replaceAndAttachPartitions(table); + case MERGE_ROWS -> mergeIntoMainTable(table); + } + } - log.info("Swapping table: '{}'", table.getMainName()); - log.info("Master table exists: '{}', skip master table: '{}'", tableExists, skipMasterTable); + /** + * Determines how staged data for the given table should be published to the main table, based on + * the update type, whether the table type supports a "latest partition", whether a main table + * already exists, and how the target database manages partitions. + */ + private TableUpdateAction resolveUpdateAction( + AnalyticsTableUpdateParams params, AnalyticsTable table) { + boolean supportsLatestPartition = table.getTableType().isLatestPartition(); - List swappedPartitions = new UniqueArrayList<>(); + // Full updates, and table types with no "latest partition" concept, always replace the table. + if (!params.isPartialUpdate() || !supportsLatestPartition) { + return TableUpdateAction.REPLACE_TABLE; + } + + // Without continuous analytics support (e.g. ClickHouse) there is no step removing updated and + // deleted rows, so a continuous update cannot be merged. The staging table holds all data on + // such engines (see DefaultAnalyticsTableService.getTablePartitions()), so replacing the whole + // main table gives correct results, as a full rebuild would. + if (params.isLatestUpdate() && !sqlBuilder.supportsContinuousAnalytics()) { + log.warn( + "Continuous (lastYears=0) update of table '{}' rebuilds the whole table on this " + + "database, which does not support incremental updates", + table.getMainName()); + return TableUpdateAction.REPLACE_TABLE; + } + // No existing table to preserve; treat as the initial build. + if (!tableExists(table.getMainName())) { + return TableUpdateAction.REPLACE_TABLE; + } + + // Postgres publishes partial updates by replacing physical (inheritance-based) partitions. if (!sqlBuilder.supportsDeclarativePartitioning()) { - table.getTablePartitions().forEach(part -> swapTable(part, part.getMainName())); - table.getTablePartitions().forEach(part -> swappedPartitions.add(part.fromStaging())); + return TableUpdateAction.REPLACE_PARTITIONS; } - if (!skipMasterTable) { - // Full replace update and main table exist, swap main table - swapTable(table, table.getMainName()); - } else { - // Incremental append update, update parent of partitions to existing main table - if (!sqlBuilder.supportsDeclarativePartitioning()) { - swappedPartitions.forEach( - partition -> swapParentTable(partition, table.getName(), table.getMainName())); - } - dropTable(table); + // Declarative-partitioning engines with no inheritance-based partition attachment (e.g. + // Doris) publish a continuous update by merging staged rows into the persistent main table. + if (params.isLatestUpdate()) { + return TableUpdateAction.MERGE_ROWS; } + + // A bounded lastYears update against a main table that already exists. Merging is not an option + // here, since removeUpdatedData() only runs for the latest-partition case and merging would + // leave stale/deleted rows behind. On declarative-partitioning engines the populate step does + // not filter by year, so the staging table holds every year and replacing the whole main table + // gives correct results. + log.warn( + "Bounded lastYears update of table '{}' rebuilds the whole table with all years on this " + + "database, use a continuous (lastYears=0) update for incremental updates", + table.getMainName()); + return TableUpdateAction.REPLACE_TABLE; + } + + private void replaceMainTable(AnalyticsTable table) { + replacePhysicalPartitions(table); + swapTable(table, table.getMainName()); + } + + private void replaceAndAttachPartitions(AnalyticsTable table) { + List
partitions = replacePhysicalPartitions(table); + + partitions.forEach( + partition -> swapParentTable(partition, table.getName(), table.getMainName())); + + dropTable(table); + } + + private void mergeIntoMainTable(AnalyticsTable table) { + String fromTable = sqlBuilder.quote(table.getName()); + jdbcTemplate.execute(sqlBuilder.insertIntoSelectFrom(table.fromStaging(), fromTable)); + dropTable(table); + } + + /** + * Swaps each of the table's physical (inheritance-based) partitions from staging to main. A no-op + * for declarative-partitioning engines, which have no separate physical partition tables. + * + * @return the swapped partitions, for a caller that needs to reparent them onto an existing main + * table. + */ + private List
replacePhysicalPartitions(AnalyticsTable table) { + if (sqlBuilder.supportsDeclarativePartitioning()) { + return List.of(); + } + + List
partitions = new UniqueArrayList<>(); + + for (AnalyticsTablePartition partition : table.getTablePartitions()) { + swapTable(partition, partition.getMainName()); + partitions.add(partition.fromStaging()); + } + + return partitions; } @Override @@ -363,10 +455,59 @@ private void swapParentTable(Table partition, String stagingMasterName, String m * @param name the table name. * @return true if a table with the given name exists. */ - private boolean tableExists(String name) { + protected boolean tableExists(String name) { return !jdbcTemplate.queryForList(sqlBuilder.tableExists(name)).isEmpty(); } + @Override + public boolean isReadyForContinuousUpdate(List tables) { + if (!sqlBuilder.requiresUniqueKeyAnalyticsTables()) { + return true; + } + + return tables.stream().allMatch(this::isReadyForContinuousUpdate); + } + + /** + * A main table created before unique-key analytics tables were introduced remains on its original + * key model until a full rebuild recreates it. A continuous update's delete step requires the + * unique key model, so it must not be attempted against such a table. + */ + private boolean isReadyForContinuousUpdate(AnalyticsTable table) { + String tableName = table.getMainName(); + + if (!tableExists(tableName)) { + // No main table yet; the initial create uses the required key model. + return true; + } + + if (hasUniqueKeyModel(tableName)) { + return true; + } + + log.error( + "Table '{}' does not use the unique key model required for continuous analytics " + + "updates on this database; run a full analytics table rebuild to recreate it with " + + "the required key model before running a continuous (lastYears=0) update", + tableName); + + return false; + } + + private boolean hasUniqueKeyModel(String tableName) { + List> rows = + jdbcTemplate.queryForList(sqlBuilder.showCreateTable(tableName)); + + if (rows.isEmpty()) { + return false; + } + + String createTableSql = + rows.get(0).values().stream().map(String::valueOf).collect(Collectors.joining(" ")); + + return createTableSql.toLowerCase().contains("unique key"); + } + // ------------------------------------------------------------------------- // Abstract methods // ------------------------------------------------------------------------- @@ -413,14 +554,16 @@ protected AnalyticsTable getRegularAnalyticsTable( AnalyticsTableUpdateParams params, List dataYears, List columns, - List sortKey) { + List sortKey, + List primaryKey) { Calendar calendar = PeriodType.getCalendar(); List years = ListUtils.mutableCopy(dataYears); Logged logged = analyticsTableSettings.getTableLogged(); Collections.sort(years); - AnalyticsTable table = new AnalyticsTable(getAnalyticsTableType(), columns, sortKey, logged); + AnalyticsTable table = + new AnalyticsTable(getAnalyticsTableType(), columns, sortKey, primaryKey, logged); for (Integer year : years) { List checks = getPartitionChecks(year, getEndDate(calendar, year)); @@ -441,11 +584,12 @@ protected AnalyticsTable getRegularAnalyticsTable( * @param columns the list of {@link AnalyticsTableColumn}. */ protected AnalyticsTable getLatestAnalyticsTable( - AnalyticsTableUpdateParams params, List columns) { + AnalyticsTableUpdateParams params, + List columns, + List primaryKey) { SystemSettings settings = settingsProvider.getCurrentSettings(); Date lastFullTableUpdate = settings.getLastSuccessfulAnalyticsTablesUpdate(); - Date lastLatestPartitionUpdate = settings.getLastSuccessfulLatestAnalyticsPartitionUpdate(); - Date lastAnyTableUpdate = DateUtils.getLatest(lastLatestPartitionUpdate, lastFullTableUpdate); + Date lastAnyTableUpdate = getLastAnyTableUpdate(settings); Assert.isTrue( lastFullTableUpdate.getTime() > 0L, @@ -455,14 +599,24 @@ protected AnalyticsTable getLatestAnalyticsTable( Date endDate = params.getStartTime(); boolean hasUpdatedData = hasUpdatedLatestData(lastAnyTableUpdate, endDate); - AnalyticsTable table = new AnalyticsTable(getAnalyticsTableType(), columns, List.of(), logged); + // Engines without a unique key on analytics tables (Postgres) wholesale-replace the latest + // partition on every run, so its window must always span back to the last full rebuild, or + // data captured by an earlier continuous run would be discarded rather than carried forward. + // Engines with a unique key (Doris) instead merge into the persistent main table with + // natural-key deduplication, so it's safe and correct to only (re)process what changed since + // the last continuous run. + Date partitionStartDate = + sqlBuilder.requiresUniqueKeyAnalyticsTables() ? lastAnyTableUpdate : lastFullTableUpdate; + + AnalyticsTable table = + new AnalyticsTable(getAnalyticsTableType(), columns, List.of(), primaryKey, logged); if (hasUpdatedData) { table.addTablePartition( - List.of(), AnalyticsTablePartition.LATEST_PARTITION, lastFullTableUpdate, endDate); + List.of(), AnalyticsTablePartition.LATEST_PARTITION, partitionStartDate, endDate); log.info( "Added latest analytics partition with start: '{}' and end: '{}'", - toLongDate(lastFullTableUpdate), + toLongDate(partitionStartDate), toLongDate(endDate)); } else { log.info( @@ -474,6 +628,28 @@ protected AnalyticsTable getLatestAnalyticsTable( return table; } + /** + * Returns the time of the last successful analytics table update of any kind, full or continuous. + * Engines with a unique key on analytics tables (Doris) start the continuous update window at + * this time, so it is read from the settings of this table type: the shared settings are also + * written when this table type was skipped or its update was aborted, and a window starting there + * would never pick up the changes that update did not publish. + * + * @param settings the current {@link SystemSettings}. + */ + protected Date getLastAnyTableUpdate(SystemSettings settings) { + if (sqlBuilder.requiresUniqueKeyAnalyticsTables()) { + AnalyticsTableType tableType = getAnalyticsTableType(); + return DateUtils.getLatest( + settings.getLastSuccessfulLatestAnalyticsPartitionUpdate(tableType), + settings.getLastSuccessfulAnalyticsTablesUpdate(tableType)); + } + + return DateUtils.getLatest( + settings.getLastSuccessfulLatestAnalyticsPartitionUpdate(), + settings.getLastSuccessfulAnalyticsTablesUpdate()); + } + /** * Executes the given SQL statement. Logs and times the operation. * diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/DefaultAnalyticsTableGenerator.java b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/DefaultAnalyticsTableGenerator.java index 7bc13c1582f6..8c329bb775e2 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/DefaultAnalyticsTableGenerator.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/DefaultAnalyticsTableGenerator.java @@ -126,18 +126,17 @@ public void generateAnalyticsTables(AnalyticsTableUpdateParams params0, JobProgr } } - Set processedTypes = EnumSet.noneOf(AnalyticsTableType.class); + Set publishedTypes = EnumSet.noneOf(AnalyticsTableType.class); for (AnalyticsTableService service : analyticsTableServices) { AnalyticsTableType tableType = service.getAnalyticsTableType(); - if (!skipTypes.contains(tableType)) { - service.create(params, progress); - processedTypes.add(tableType); + if (!skipTypes.contains(tableType) && service.create(params, progress)) { + publishedTypes.add(tableType); } } progress.startingStage("Updating system settings"); - progress.runStage(() -> updateLastSuccessfulSystemSettings(params, clock, processedTypes)); + progress.runStage(() -> updateLastSuccessfulSystemSettings(params, clock, publishedTypes)); progress.startingStage("Invalidate analytics caches", SKIP_STAGE); progress.runStage(analyticsCache::invalidateAll); @@ -151,31 +150,32 @@ public void generateAnalyticsTables(AnalyticsTableUpdateParams params0, JobProgr * * @param params the {@link AnalyticsTableUpdateParams}. * @param clock the {@link Clock}. - * @param processedTypes the {@link AnalyticsTableType} values actually processed in this run, - * i.e. not in {@link AnalyticsTableUpdateParams#getSkipTableTypes()}. + * @param publishedTypes the {@link AnalyticsTableType} values published in this run, i.e. not in + * {@link AnalyticsTableUpdateParams#getSkipTableTypes()} and not cancelled or aborted, see + * {@link AnalyticsTableService#create}. */ private void updateLastSuccessfulSystemSettings( - AnalyticsTableUpdateParams params, Clock clock, Set processedTypes) { + AnalyticsTableUpdateParams params, Clock clock, Set publishedTypes) { if (params.isLatestUpdate()) { settingsService.put("keyLastSuccessfulLatestAnalyticsPartitionUpdate", params.getStartTime()); settingsService.put("keyLastSuccessfulLatestAnalyticsPartitionRuntime", clock.time()); updatePerTypeSuccessfulSystemSettings( SystemSettings::keyLastSuccessfulLatestAnalyticsPartitionUpdate, params.getStartTime(), - processedTypes); + publishedTypes); } else { settingsService.put("keyLastSuccessfulAnalyticsTablesUpdate", params.getStartTime()); settingsService.put("keyLastSuccessfulAnalyticsTablesRuntime", clock.time()); updatePerTypeSuccessfulSystemSettings( SystemSettings::keyLastSuccessfulAnalyticsTablesUpdate, params.getStartTime(), - processedTypes); + publishedTypes); } } /** * Stamps the per-{@link AnalyticsTableType} counterpart of the shared "last successful update" - * clock, for every processed type that has one registered (see {@link + * clock, for every published type that has one registered (see {@link * AnalyticsTableType#isLatestPartition()}). A type without a registered key is skipped rather * than attempted, since {@link SystemSettingsService#put} silently drops writes for keys not in * {@link SystemSettings#keysWithDefaults()}. @@ -183,8 +183,8 @@ private void updateLastSuccessfulSystemSettings( private void updatePerTypeSuccessfulSystemSettings( Function keyFunction, Date startTime, - Set processedTypes) { - processedTypes.stream() + Set publishedTypes) { + publishedTypes.stream() .filter(AnalyticsTableType::isLatestPartition) .forEach(type -> settingsService.put(keyFunction.apply(type), startTime)); } diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/DefaultAnalyticsTableService.java b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/DefaultAnalyticsTableService.java index 0bb165241ab4..e64b93fe56f6 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/DefaultAnalyticsTableService.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/DefaultAnalyticsTableService.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2004-2022, University of Oslo + * Copyright (c) 2004-2026, University of Oslo * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -83,7 +83,7 @@ public AnalyticsTableType getAnalyticsTableType() { } @Override - public void create(AnalyticsTableUpdateParams params, JobProgress progress) { + public boolean create(AnalyticsTableUpdateParams params, JobProgress progress) { final int parallelJobs = getParallelJobs(); int tableUpdates = 0; @@ -102,8 +102,12 @@ public void create(AnalyticsTableUpdateParams params, JobProgress progress) { boolean validState = tableManager.validState(); progress.completedStage("Validated analytics tables with outcome: {}", validState); - if (!validState || progress.isCancelled()) { - return; + if (progress.isCancelled()) { + return false; + } + + if (!validState) { + return true; } List tables = tableManager.getAnalyticsTables(params); @@ -112,7 +116,15 @@ public void create(AnalyticsTableUpdateParams params, JobProgress progress) { clock.logTime("Table update aborted, nothing to update: '{}'", tableType.getTableName()); progress.startingStage("Table update of type: '{}'", tableType); progress.completedStage("Table updated aborted, no table or partitions to be updated"); - return; + return true; + } + + boolean continuousUpdate = params.isLatestUpdate() && sqlBuilder.supportsContinuousAnalytics(); + + // Checked before any staging work: a table type that is not ready would otherwise populate + // its staging tables in full on every scheduled run, only to throw them away. + if (continuousUpdate && !checkContinuousUpdateReadiness(tables, tableType, progress, clock)) { + return false; } clock.logTime( @@ -132,7 +144,7 @@ public void create(AnalyticsTableUpdateParams params, JobProgress progress) { createTables(tables, progress); clock.logTime("Created analytics tables"); - List partitions = getTablePartitions(tables); + List partitions = getTablePartitions(tables, params.isLatestUpdate()); int partitionSize = partitions.size(); progress.startingStage( @@ -169,14 +181,15 @@ public void create(AnalyticsTableUpdateParams params, JobProgress progress) { clock.logTime("Analyzed tables"); } - if (params.isLatestUpdate()) { - progress.startingStage( - format("Removing updated and deleted data: '{}'", tableType), SKIP_STAGE); - progress.runStage(() -> tableManager.removeUpdatedData(tables)); - clock.logTime("Removed updated and deleted data"); + List publishedTables = tables; + + if (continuousUpdate) { + publishedTables = removeUpdatedData(tables, tableType, progress, clock); } - swapTables(params, tables, progress); + if (!publishedTables.isEmpty()) { + swapTables(params, publishedTables, progress); + } if (params.isPartialUpdate() && !params.isLatestUpdate()) { progress.startingStage("Removing latest partition overlap: '{}'", tableType); @@ -185,6 +198,9 @@ public void create(AnalyticsTableUpdateParams params, JobProgress progress) { } clock.logTime("Table update done: '{}'", tableType.getTableName()); + // Not complete if a table's staged data could not be published, so the update window is not + // advanced and the next continuous run covers its changes again + return publishedTables.size() == tables.size(); } @Override @@ -321,6 +337,115 @@ private void analyzeTables(List tables, JobProgress progress) { getParallelJobs(), tables, Table::getName, tableManager::analyzeTable); } + /** + * Checks that the main analytics tables can take a continuous update. The check is cheap, so it + * runs before any staging tables are created and populated. + * + * @param tables the list of {@link AnalyticsTable}. + * @param tableType the {@link AnalyticsTableType}. + * @param progress the {@link JobProgress}. + * @param clock the {@link Clock}. + * @return true if the continuous update can proceed, false if it must be aborted. + */ + private boolean checkContinuousUpdateReadiness( + List tables, + AnalyticsTableType tableType, + JobProgress progress, + Clock clock) { + progress.startingStage( + format("Validating continuous update readiness: '{}'", tableType), SKIP_STAGE); + + if (!tableManager.isReadyForContinuousUpdate(tables)) { + // A main table predating unique-key analytics tables (or otherwise not in a state that + // supports a continuous update) must not have the delete step attempted against it, see + // isReadyForContinuousUpdate() for detail. The stage is reported as failed so the reason + // shows in the job log, while the remaining table types still get updated. + progress.failedStage( + format( + "Continuous update skipped for '{}': the analytics table was created before " + + "continuous updates were supported on this database. Run a full analytics " + + "table rebuild before running a continuous (lastYears=0) update", + tableType)); + clock.logTime("Continuous analytics update aborted, not ready: '{}'", tableType); + return false; + } + + progress.completedStage("Validated continuous update readiness: '{}'", tableType); + return true; + } + + /** + * Removes the rows which the staged data replaces from the main analytics tables. + * + *

On databases which require unique-key analytics tables (Doris), each table is handled on its + * own: a table whose delete fails keeps its staged data unpublished, while the other tables are + * published. Merging a window again is idempotent there, so the next run, which covers the same + * window again, publishes the failed table's changes once its delete succeeds. + * + *

On other databases the tables are swapped in after the delete whatever its outcome, as + * before continuous updates were supported on unique-key databases. + * + * @param tables the list of {@link AnalyticsTable}. + * @param tableType the {@link AnalyticsTableType}. + * @param progress the {@link JobProgress}. + * @param clock the {@link Clock}. + * @return the tables whose staged data can be published. + */ + private List removeUpdatedData( + List tables, + AnalyticsTableType tableType, + JobProgress progress, + Clock clock) { + progress.startingStage( + format("Removing updated and deleted data: '{}'", tableType), SKIP_STAGE); + + if (!sqlBuilder.requiresUniqueKeyAnalyticsTables()) { + progress.runStage(() -> tableManager.removeUpdatedData(tables)); + clock.logTime("Removed updated and deleted data"); + return tables; + } + + List publishedTables = new ArrayList<>(); + List failedTables = new ArrayList<>(); + + for (AnalyticsTable table : tables) { + // A table without a main table yet (e.g. the event table of a program created after the + // last full rebuild) has no stale rows to remove. Its staging table becomes the main table. + if (!tableManager.mainTableExists(table)) { + publishedTables.add(table); + continue; + } + + try { + tableManager.removeUpdatedData(List.of(table)); + publishedTables.add(table); + } catch (RuntimeException ex) { + // Swapping in the staged data without having removed the rows it replaces would leave + // duplicate or orphaned rows, so this table's staged data is not published + log.error( + "Failed to remove updated and deleted data from '{}', its staged data is not published", + table.getMainName(), + ex); + failedTables.add(table.getMainName()); + } + } + + clock.logTime("Removed updated and deleted data"); + + if (failedTables.isEmpty()) { + progress.completedStage("Removed updated and deleted data: '{}'", tableType); + } else { + progress.failedStage( + format( + "Failed to remove updated and deleted data from {}. Their staged data was not " + + "published and the next continuous update covers it again. If this keeps " + + "failing, a full analytics table rebuild may be required", + failedTables)); + } + + return publishedTables; + } + /** * Swaps the given analytics tables. * @@ -347,12 +472,27 @@ private void swapTables( * @param tables the list of {@link AnalyticsTable}. * @return a list of {@link AnalyticsTablePartition}. */ - List getTablePartitions(List tables) { + List getTablePartitions( + List tables, boolean isLatestUpdate) { List partitions = new ArrayList<>(); for (AnalyticsTable table : tables) { if (table.hasTablePartitions() && !sqlBuilder.supportsDeclarativePartitioning()) { + // Each partition is its own physical table, so its own real date range and name apply. partitions.addAll(table.getTablePartitions()); + } else if (table.hasTablePartitions() + && isLatestUpdate + && sqlBuilder.supportsContinuousAnalytics()) { + // A single physical table serves every logical partition on this engine (CREATE only + // ever builds the master table's name), so the continuous/latest-update partition must + // target that name - while still carrying its own real date range, needed to correctly + // scope the populate window. Engines without continuous analytics support (ClickHouse) + // fall through to the fake partition below, so the staging table holds all data and can + // replace the whole main table. + AnalyticsTablePartition latest = table.getTablePartitions().get(0); + partitions.add( + new AnalyticsTablePartition( + table, latest.getYear(), latest.getStartDate(), latest.getEndDate())); } else { // Fake partition representing the master table partitions.add(new AnalyticsTablePartition(table)); diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManager.java b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManager.java index 6630c95bb721..8ed24179dc23 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManager.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManager.java @@ -76,6 +76,7 @@ import org.hisp.dhis.common.ValueType; import org.hisp.dhis.configuration.ConfigurationService; import org.hisp.dhis.dataapproval.DataApprovalLevelService; +import org.hisp.dhis.db.model.Column; import org.hisp.dhis.db.model.Table; import org.hisp.dhis.db.sql.SqlBuilder; import org.hisp.dhis.organisationunit.OrganisationUnitLevel; @@ -168,6 +169,8 @@ public class JdbcAnalyticsTableManager extends AbstractJdbcTableManager { private static final List SORT_KEY = List.of("dx", "co"); + private static final List PRIMARY_KEY = List.of("id", "year"); + public JdbcAnalyticsTableManager( IdentifiableObjectManager idObjectManager, OrganisationUnitService organisationUnitService, @@ -210,10 +213,14 @@ public AnalyticsTableType getAnalyticsTableType() { @Override @Transactional public List getAnalyticsTables(AnalyticsTableUpdateParams params) { + List primaryKey = + sqlBuilder.requiresUniqueKeyAnalyticsTables() ? PRIMARY_KEY : List.of(); + AnalyticsTable table = params.isLatestUpdate() - ? getLatestAnalyticsTable(params, getColumns(params)) - : getRegularAnalyticsTable(params, getDataYears(params), getColumns(params), SORT_KEY); + ? getLatestAnalyticsTable(params, getColumns(params), primaryKey) + : getRegularAnalyticsTable( + params, getDataYears(params), getColumns(params), SORT_KEY, primaryKey); return table.hasTablePartitions() ? List.of(table) : List.of(); } @@ -252,6 +259,86 @@ public void preCreateTables(AnalyticsTableUpdateParams params) { @Override public void removeUpdatedData(List tables) { AnalyticsTablePartition partition = getLatestTablePartition(tables); + + if (sqlBuilder.requiresUniqueKeyAnalyticsTables()) { + removeUpdatedDataViaStagingKeys(partition); + } else { + removeUpdatedDataViaSubquery(partition); + } + } + + /** + * Removes updated and deleted data on databases which require a unique key for analytics tables + * (Doris). The equivalent single {@code delete ... using } statement is avoided + * here because it joins a live federated table (the operational database's {@code datavalue}) + * against local tables inside a {@code delete}, which is dramatically slower than the same join + * shape used as a plain {@code select} (as {@link #populateTable} already does). Instead, the + * matching rows' natural key column values are first materialized into a small native staging + * table via that proven-fast federated {@code select}, and the actual delete runs entirely + * locally against that staging table. + */ + private void removeUpdatedDataViaStagingKeys(AnalyticsTablePartition partition) { + String keysTableName = getAnalyticsTableType().getTableName() + "_delete_keys"; + Table keysTable = + new Table( + keysTableName, + List.of( + new Column("dx", CHARACTER_11), + new Column("co", CHARACTER_11), + new Column("ao", CHARACTER_11), + new Column("pe", TEXT), + new Column("ou", CHARACTER_11), + new Column("year", INTEGER)), + List.of()); + + invokeTimeAndLog( + sqlBuilder.dropTableIfExistsCascade(keysTableName), "Drop delete-keys staging table (pre)"); + + invokeTimeAndLog(sqlBuilder.createTable(keysTable), "Create delete-keys staging table"); + + String insertSql = + replaceQualify( + sqlBuilder, + """ + insert into ${keysTableName} (dx,co,ao,pe,ou,year) \ + select des.dataelementuid, dcs.categoryoptioncombouid, acs.categoryoptioncombouid, ps.iso, ous.organisationunituid, ps.year \ + from ${datavalue} dv \ + inner join analytics_rs_dataelementstructure des on dv.dataelementid=des.dataelementid \ + inner join analytics_rs_periodstructure ps on dv.periodid=ps.periodid \ + inner join analytics_rs_orgunitstructure ous on dv.sourceid=ous.organisationunitid \ + inner join analytics_rs_categorystructure dcs on dv.categoryoptioncomboid=dcs.categoryoptioncomboid \ + inner join analytics_rs_categorystructure acs on dv.attributeoptioncomboid=acs.categoryoptioncomboid \ + where dv.lastupdated >= '${startDate}' and dv.lastupdated < '${endDate}';""", + Map.of( + "keysTableName", quote(keysTableName), + "startDate", toLongDate(partition.getStartDate()), + "endDate", toLongDate(partition.getEndDate()))); + + invokeTimeAndLog(insertSql, "Populate delete-keys staging table"); + + String deleteSql = + replace( + """ + delete from ${tableName} ax \ + using ${keysTableName} k \ + where ax.dx=k.dx and ax.co=k.co and ax.ao=k.ao and ax.pe=k.pe and ax.ou=k.ou and ax.year=k.year;""", + Map.of( + "tableName", quote(getAnalyticsTableType().getTableName()), + "keysTableName", quote(keysTableName))); + + invokeTimeAndLog(deleteSql, "Remove updated data values"); + + invokeTimeAndLog( + sqlBuilder.dropTableIfExistsCascade(keysTableName), + "Drop delete-keys staging table (post)"); + } + + /** + * Removes updated and deleted data on databases which do not require a unique key for analytics + * tables (Postgres, ClickHouse). Runs entirely against local tables, so the federated-join + * problem {@link #removeUpdatedDataViaStagingKeys} works around does not apply here. + */ + private void removeUpdatedDataViaSubquery(AnalyticsTablePartition partition) { String sql = replaceQualify( sqlBuilder, @@ -531,7 +618,7 @@ private List getColumns(AnalyticsTableUpdateParams params) columns.add( AnalyticsTableColumn.builder() .name("id") - .dataType(TEXT) + .dataType(sqlBuilder.requiresUniqueKeyAnalyticsTables() ? VARCHAR_255 : TEXT) .selectExpression(idColAlias) .build()); columns.addAll(getDataElementGroupSetColumns()); diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcCompletenessTableManager.java b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcCompletenessTableManager.java index 3063acadabbb..d4907d3d32b6 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcCompletenessTableManager.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcCompletenessTableManager.java @@ -41,6 +41,7 @@ import static org.hisp.dhis.db.model.DataType.INTEGER; import static org.hisp.dhis.db.model.DataType.TEXT; import static org.hisp.dhis.db.model.DataType.TIMESTAMP; +import static org.hisp.dhis.db.model.DataType.VARCHAR_255; import static org.hisp.dhis.db.model.constraint.Nullable.NOT_NULL; import static org.hisp.dhis.util.DateUtils.SECONDS_PER_DAY; import static org.hisp.dhis.util.DateUtils.toLongDate; @@ -96,6 +97,8 @@ public class JdbcCompletenessTableManager extends AbstractJdbcTableManager { private static final List SORT_KEY = List.of("dx"); + private static final List PRIMARY_KEY = List.of("id", "year"); + public JdbcCompletenessTableManager( IdentifiableObjectManager idObjectManager, OrganisationUnitService organisationUnitService, @@ -134,10 +137,14 @@ public AnalyticsTableType getAnalyticsTableType() { @Override @Transactional public List getAnalyticsTables(AnalyticsTableUpdateParams params) { + List primaryKey = + sqlBuilder.requiresUniqueKeyAnalyticsTables() ? PRIMARY_KEY : List.of(); + AnalyticsTable table = params.isLatestUpdate() - ? getLatestAnalyticsTable(params, getColumns()) - : getRegularAnalyticsTable(params, getDataYears(params), getColumns(), SORT_KEY); + ? getLatestAnalyticsTable(params, getColumns(), primaryKey) + : getRegularAnalyticsTable( + params, getDataYears(params), getColumns(), SORT_KEY, primaryKey); return table.hasTablePartitions() ? List.of(table) : List.of(); } @@ -172,23 +179,39 @@ public boolean hasUpdatedLatestData(Date startDate, Date endDate) { public void removeUpdatedData(List tables) { AnalyticsTablePartition partition = getLatestTablePartition(tables); String sql = - replaceQualify( - sqlBuilder, - """ - delete from ${tableName} ax \ - where ax.id in ( \ - select concat(ds.uid,'-',ps.iso,'-',ous.organisationunituid,'-',acs.categoryoptioncombouid) as id \ - from ${completedatasetregistration} cdr \ - inner join ${dataset} ds on cdr.datasetid=ds.datasetid \ - inner join analytics_rs_periodstructure ps on cdr.periodid=ps.periodid \ - inner join analytics_rs_orgunitstructure ous on cdr.sourceid=ous.organisationunitid \ - inner join analytics_rs_categorystructure acs on cdr.attributeoptioncomboid=acs.categoryoptioncomboid \ - where cdr.lastupdated >= '${startDate}' \ - and cdr.lastupdated < '${endDate}');""", - Map.of( - "tableName", quote(getAnalyticsTableType().getTableName()), - "startDate", toLongDate(partition.getStartDate()), - "endDate", toLongDate(partition.getEndDate()))); + sqlBuilder.requiresUniqueKeyAnalyticsTables() + ? replaceQualify( + sqlBuilder, + """ + delete from ${tableName} ax \ + using ${completedatasetregistration} cdr \ + inner join ${dataset} ds on cdr.datasetid=ds.datasetid \ + inner join analytics_rs_periodstructure ps on cdr.periodid=ps.periodid \ + inner join analytics_rs_orgunitstructure ous on cdr.sourceid=ous.organisationunitid \ + inner join analytics_rs_categorystructure acs on cdr.attributeoptioncomboid=acs.categoryoptioncomboid \ + where ax.id = concat(ds.uid,'-',ps.iso,'-',ous.organisationunituid,'-',acs.categoryoptioncombouid) \ + and cdr.lastupdated >= '${startDate}' and cdr.lastupdated < '${endDate}';""", + Map.of( + "tableName", quote(getAnalyticsTableType().getTableName()), + "startDate", toLongDate(partition.getStartDate()), + "endDate", toLongDate(partition.getEndDate()))) + : replaceQualify( + sqlBuilder, + """ + delete from ${tableName} ax \ + where ax.id in ( \ + select concat(ds.uid,'-',ps.iso,'-',ous.organisationunituid,'-',acs.categoryoptioncombouid) as id \ + from ${completedatasetregistration} cdr \ + inner join ${dataset} ds on cdr.datasetid=ds.datasetid \ + inner join analytics_rs_periodstructure ps on cdr.periodid=ps.periodid \ + inner join analytics_rs_orgunitstructure ous on cdr.sourceid=ous.organisationunitid \ + inner join analytics_rs_categorystructure acs on cdr.attributeoptioncomboid=acs.categoryoptioncomboid \ + where cdr.lastupdated >= '${startDate}' \ + and cdr.lastupdated < '${endDate}');""", + Map.of( + "tableName", quote(getAnalyticsTableType().getTableName()), + "startDate", toLongDate(partition.getStartDate()), + "endDate", toLongDate(partition.getEndDate()))); invokeTimeAndLog(sql, "Remove updated data values"); } @@ -267,7 +290,7 @@ private List getColumns() { columns.add( AnalyticsTableColumn.builder() .name("id") - .dataType(TEXT) + .dataType(sqlBuilder.requiresUniqueKeyAnalyticsTables() ? VARCHAR_255 : TEXT) .selectExpression(idColAlias) .build()); columns.addAll(getOrganisationUnitGroupSetColumns()); diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcEventAnalyticsTableManager.java b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcEventAnalyticsTableManager.java index c78623f8c4bb..b2a885a57634 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcEventAnalyticsTableManager.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcEventAnalyticsTableManager.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2004-2022, University of Oslo + * Copyright (c) 2004-2026, University of Oslo * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -81,8 +81,10 @@ import org.hisp.dhis.configuration.ConfigurationService; import org.hisp.dhis.dataapproval.DataApprovalLevelService; import org.hisp.dhis.dataelement.DataElement; +import org.hisp.dhis.db.model.Column; import org.hisp.dhis.db.model.DataType; import org.hisp.dhis.db.model.Logged; +import org.hisp.dhis.db.model.Table; import org.hisp.dhis.db.sql.SqlBuilder; import org.hisp.dhis.legend.Legend; import org.hisp.dhis.legend.LegendSet; @@ -95,7 +97,6 @@ import org.hisp.dhis.setting.SystemSettings; import org.hisp.dhis.setting.SystemSettingsProvider; import org.hisp.dhis.trackedentity.TrackedEntityAttribute; -import org.hisp.dhis.util.DateUtils; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Service; @@ -111,6 +112,8 @@ public class JdbcEventAnalyticsTableManager extends AbstractEventJdbcTableManage static final String[] EXPORTABLE_EVENT_STATUSES = {"'COMPLETED'", "'ACTIVE'", "'SCHEDULE'"}; + private static final List PRIMARY_KEY = List.of("event", "year"); + public JdbcEventAnalyticsTableManager( IdentifiableObjectManager idObjectManager, OrganisationUnitService organisationUnitService, @@ -195,7 +198,12 @@ private List getRegularAnalyticsTables( Collections.sort(yearsForPartitionTables); AnalyticsTable table = - new AnalyticsTable(getAnalyticsTableType(), getColumns(program), logged, program); + new AnalyticsTable( + getAnalyticsTableType(), + getColumns(program), + sqlBuilder.requiresUniqueKeyAnalyticsTables() ? PRIMARY_KEY : List.of(), + logged, + program); for (Integer year : yearsForPartitionTables) { List checks = getPartitionChecks(year, PartitionUtils.getEndDate(calendar, year)); @@ -225,14 +233,20 @@ private List getRegularAnalyticsTables( private List getLatestAnalyticsTables(AnalyticsTableUpdateParams params) { SystemSettings settings = settingsProvider.getCurrentSettings(); Date lastFullTableUpdate = settings.getLastSuccessfulAnalyticsTablesUpdate(); - Date lastLatestPartitionUpdate = settings.getLastSuccessfulLatestAnalyticsPartitionUpdate(); - Date lastAnyTableUpdate = DateUtils.getLatest(lastLatestPartitionUpdate, lastFullTableUpdate); + Date lastAnyTableUpdate = getLastAnyTableUpdate(settings); Assert.isTrue( lastFullTableUpdate.getTime() > 0L, "A full analytics table update process must be run prior to a latest partition update"); - Date startDate = lastFullTableUpdate; + // Engines without a unique key on analytics tables (Postgres) wholesale-replace the latest + // partition on every run, so its window must always span back to the last full rebuild, or + // data captured by an earlier continuous run would be discarded rather than carried forward. + // Engines with a unique key (Doris) instead merge into the persistent main table with + // natural-key deduplication, so it's safe and correct to only (re)process what changed since + // the last continuous run. + Date startDate = + sqlBuilder.requiresUniqueKeyAnalyticsTables() ? lastAnyTableUpdate : lastFullTableUpdate; Date endDate = params.getStartTime(); List tables = new ArrayList<>(); @@ -247,25 +261,41 @@ private List getLatestAnalyticsTables(AnalyticsTableUpdateParams : idObjectManager.getAllNoAcl(Program.class); for (Program program : programs) { - boolean hasUpdatedData = hasUpdatedLatestData(lastAnyTableUpdate, endDate, program); + // On databases which merge the latest partition into the main table (Doris), a program + // without a main table yet (e.g. created after the last full rebuild) gets all of its data, + // since its staging table becomes the main table and later runs only merge into it. A window + // starting at the last update would leave out its events last updated before then. Other + // databases replace the latest partition with the same window on every run. + boolean fullHistory = + sqlBuilder.requiresUniqueKeyAnalyticsTables() + && !tableExists(AnalyticsTable.getTableName(getAnalyticsTableType(), program)); + Date programStartDate = fullHistory ? new Date(0L) : startDate; + Date updatedSince = fullHistory ? new Date(0L) : lastAnyTableUpdate; + + boolean hasUpdatedData = hasUpdatedLatestData(updatedSince, endDate, program); if (hasUpdatedData) { AnalyticsTable table = - new AnalyticsTable(getAnalyticsTableType(), getColumns(program), logged, program); + new AnalyticsTable( + getAnalyticsTableType(), + getColumns(program), + sqlBuilder.requiresUniqueKeyAnalyticsTables() ? PRIMARY_KEY : List.of(), + logged, + program); table.addTablePartition( - List.of(), AnalyticsTablePartition.LATEST_PARTITION, startDate, endDate); + List.of(), AnalyticsTablePartition.LATEST_PARTITION, programStartDate, endDate); tables.add(table); log.info( "Added latest event analytics partition for program: '{}', start: '{}' and end: '{}'", program.getUid(), - toLongDate(startDate), + toLongDate(programStartDate), toLongDate(endDate)); } else { log.info( "No updated latest event data found for program: '{}', start: '{}' and end: '{}", program.getUid(), - toLongDate(lastAnyTableUpdate), + toLongDate(updatedSince), toLongDate(endDate)); } } @@ -361,61 +391,204 @@ private boolean hasNewTrackerEvents(Date startDate, Date endDate, Program progra public void removeUpdatedData(List tables) { for (AnalyticsTable table : tables) { AnalyticsTablePartition partition = table.getLatestTablePartition(); - String sql = null; Program program = table.getProgram(); - if (table.getProgram().isRegistration()) { + if (sqlBuilder.requiresUniqueKeyAnalyticsTables()) { + removeUpdatedDataViaStagingKeys(table, program, partition); + } else { + String sql = getSubquerySql(table, program, partition); + + if (isNotBlank(sql)) { + invokeTimeAndLog(sql, "Remove updated events for table: '{}'", table.getMainName()); + } + } + } + } + + /** + * Removes updated and deleted events on databases which do not require a unique key for analytics + * tables (Postgres, ClickHouse). Runs entirely against local tables, so the federated-join + * problem {@link #removeUpdatedDataViaStagingKeys} works around does not apply here. + */ + private String getSubquerySql( + AnalyticsTable table, Program program, AnalyticsTablePartition partition) { + + String sql = null; + if (table.getProgram().isRegistration()) { + sql = + replaceQualify( + sqlBuilder, + """ + delete from ${tableName} ax \ + where ax.event in ( \ + select ev.uid \ + from ${trackerevent} ev \ + inner join ${enrollment} en on ev.enrollmentid=en.enrollmentid \ + where en.programid = ${programId} \ + and ev.lastupdated >= '${startDate}' \ + and ev.lastupdated < '${endDate}');""", + Map.of( + "tableName", sqlBuilder.qualifyTable(table.getMainName()), + "programId", String.valueOf(program.getId()), + "startDate", toLongDate(partition.getStartDate()), + "endDate", toLongDate(partition.getEndDate()))); + } else { + if (isEmpty(program.getProgramStages()) || program.getProgramStages().size() > 1) { + log.warn("Single event program {}, must have one stage", program.getUid()); + } else { + String programStageId = + String.valueOf( + program.isSingleProgramStage() + ? program.getProgramStages().stream().toList().get(0).getId() + : EMPTY); sql = replaceQualify( sqlBuilder, """ - delete from ${tableName} ax \ - where ax.event in ( \ - select ev.uid \ - from ${trackerevent} ev \ - inner join ${enrollment} en on ev.enrollmentid=en.enrollmentid \ - where en.programid = ${programId} \ - and ev.lastupdated >= '${startDate}' \ - and ev.lastupdated < '${endDate}');""", + delete from ${tableName} ax \ + where ax.event in ( \ + select ev.uid \ + from ${singleevent} ev \ + where ev.programstageid = ${programStageId} \ + and ev.lastupdated >= '${startDate}' \ + and ev.lastupdated < '${endDate}');""", Map.of( "tableName", sqlBuilder.qualifyTable(table.getMainName()), - "programId", String.valueOf(program.getId()), + "programStageId", String.valueOf(programStageId), "startDate", toLongDate(partition.getStartDate()), "endDate", toLongDate(partition.getEndDate()))); - } else { - if (isEmpty(program.getProgramStages()) || program.getProgramStages().size() > 1) { - log.warn("Single event program {}, must have one stage", program.getUid()); - } else { - String programStageId = - String.valueOf( - program.isSingleProgramStage() - ? program.getProgramStages().stream().toList().get(0).getId() - : EMPTY); - sql = - replaceQualify( - sqlBuilder, - """ - delete from ${tableName} ax \ - where ax.event in ( \ - select ev.uid \ - from ${singleevent} ev \ - where ev.programstageid = ${programStageId} \ - and ev.lastupdated >= '${startDate}' \ - and ev.lastupdated < '${endDate}');""", - Map.of( - "tableName", sqlBuilder.qualifyTable(table.getMainName()), - "programStageId", String.valueOf(programStageId), - "startDate", toLongDate(partition.getStartDate()), - "endDate", toLongDate(partition.getEndDate()))); - } } + } + return sql; + } - if (isNotBlank(sql)) { - invokeTimeAndLog(sql, "Remove updated events for table: '{}'", table.getMainName()); - } + /** + * Removes updated and deleted events on databases which require a unique key for analytics tables + * (Doris). The equivalent single {@code delete ... using } statement is avoided + * here because it joins a live federated table (the operational database's {@code + * trackerevent}/{@code singleevent}) against local tables inside a {@code delete}, which is + * dramatically slower than the same join shape used as a plain {@code select} (as {@link + * #populateTable} already does). Instead, the matching rows' {@code event} UID is first + * materialized into a small native staging table via that proven-fast federated {@code select}, + * and the actual delete runs entirely locally against that staging table. Unlike the aggregate + * ({@code DATA_VALUE}) case, an analytics event row is identified by its {@code event} UID alone + * (one physical table per program, one row per event). {@link #PRIMARY_KEY} also includes {@code + * year} only because Doris requires the partition column in a unique key. The staging table + * therefore carries just {@code event}, and the delete matches on it across all years, so an + * event whose date moved to a different year is still removed. + */ + private void removeUpdatedDataViaStagingKeys( + AnalyticsTable table, Program program, AnalyticsTablePartition partition) { + if (program.isRegistration()) { + removeUpdatedTrackerEventsViaStagingKeys(table, program, partition); + } else if (isEmpty(program.getProgramStages()) || program.getProgramStages().size() > 1) { + log.warn("Single event program {}, must have one stage", program.getUid()); + } else { + removeUpdatedSingleEventsViaStagingKeys(table, program, partition); } } + private void removeUpdatedTrackerEventsViaStagingKeys( + AnalyticsTable table, Program program, AnalyticsTablePartition partition) { + String keysTableName = getDeleteKeysTableName(); + + invokeTimeAndLog( + sqlBuilder.dropTableIfExistsCascade(keysTableName), "Drop delete-keys staging table (pre)"); + + invokeTimeAndLog( + sqlBuilder.createTable(newDeleteKeysTable(keysTableName)), + "Create delete-keys staging table"); + + String insertSql = + replaceQualify( + sqlBuilder, + """ + insert into ${keysTableName} (event) \ + select ev.uid \ + from ${trackerevent} ev \ + inner join ${enrollment} en on ev.enrollmentid=en.enrollmentid \ + where en.programid = ${programId} \ + and ev.lastupdated >= '${startDate}' and ev.lastupdated < '${endDate}';""", + Map.of( + "keysTableName", quote(keysTableName), + "programId", String.valueOf(program.getId()), + "startDate", toLongDate(partition.getStartDate()), + "endDate", toLongDate(partition.getEndDate()))); + + invokeTimeAndLog(insertSql, "Populate delete-keys staging table"); + + invokeTimeAndLog( + deleteFromMainTableUsingKeysSql(table, keysTableName), + "Remove updated events for table: '{}'", + table.getMainName()); + + invokeTimeAndLog( + sqlBuilder.dropTableIfExistsCascade(keysTableName), + "Drop delete-keys staging table (post)"); + } + + private void removeUpdatedSingleEventsViaStagingKeys( + AnalyticsTable table, Program program, AnalyticsTablePartition partition) { + String programStageId = + String.valueOf( + program.isSingleProgramStage() + ? program.getProgramStages().stream().toList().get(0).getId() + : EMPTY); + String keysTableName = getDeleteKeysTableName(); + + invokeTimeAndLog( + sqlBuilder.dropTableIfExistsCascade(keysTableName), "Drop delete-keys staging table (pre)"); + + invokeTimeAndLog( + sqlBuilder.createTable(newDeleteKeysTable(keysTableName)), + "Create delete-keys staging table"); + + String insertSql = + replaceQualify( + sqlBuilder, + """ + insert into ${keysTableName} (event) \ + select ev.uid \ + from ${singleevent} ev \ + where ev.programstageid = ${programStageId} \ + and ev.lastupdated >= '${startDate}' and ev.lastupdated < '${endDate}';""", + Map.of( + "keysTableName", quote(keysTableName), + "programStageId", programStageId, + "startDate", toLongDate(partition.getStartDate()), + "endDate", toLongDate(partition.getEndDate()))); + + invokeTimeAndLog(insertSql, "Populate delete-keys staging table"); + + invokeTimeAndLog( + deleteFromMainTableUsingKeysSql(table, keysTableName), + "Remove updated events for table: '{}'", + table.getMainName()); + + invokeTimeAndLog( + sqlBuilder.dropTableIfExistsCascade(keysTableName), + "Drop delete-keys staging table (post)"); + } + + private String deleteFromMainTableUsingKeysSql(AnalyticsTable table, String keysTableName) { + return replace( + """ + delete from ${tableName} ax \ + using ${keysTableName} k \ + where ax.event=k.event;""", + Map.of( + "tableName", quote(table.getMainName()), + "keysTableName", quote(keysTableName))); + } + + private String getDeleteKeysTableName() { + return getAnalyticsTableType().getTableName() + "_delete_keys"; + } + + private Table newDeleteKeysTable(String keysTableName) { + return new Table(keysTableName, List.of(new Column("event", CHARACTER_11)), List.of()); + } + @Override protected List getPartitionChecks(Integer year, Date endDate) { Objects.requireNonNull(year); diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcValidationResultTableManager.java b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcValidationResultTableManager.java index ee1874b254d2..b5afcd31de4a 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcValidationResultTableManager.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/JdbcValidationResultTableManager.java @@ -144,7 +144,8 @@ public List getAnalyticsTables(AnalyticsTableUpdateParams params params.isLatestUpdate() ? new AnalyticsTable( AnalyticsTableType.VALIDATION_RESULT, getColumns(), List.of(), Logged.LOGGED) - : getRegularAnalyticsTable(params, getDataYears(params), getColumns(), SORT_KEY); + : getRegularAnalyticsTable( + params, getDataYears(params), getColumns(), SORT_KEY, List.of()); return table.hasTablePartitions() ? List.of(table) : List.of(); } diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/model/AnalyticsTable.java b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/model/AnalyticsTable.java index 4723736b5df0..aef7b396b34b 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/model/AnalyticsTable.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/model/AnalyticsTable.java @@ -95,6 +95,34 @@ public AnalyticsTable( this.analyticsTableColumns = columns; } + /** + * Constructor. Sets the name to represent a staging table. Use when the table requires both a + * sort key and a real primary key (e.g. Doris continuous analytics, which needs a unique key + * column for deduplication and DELETE support). + * + * @param tableType the {@link AnalyticsTableType}. + * @param columns the list of {@link Column}. + * @param sortKey the sort key. + * @param primaryKey the primary key. + * @param logged the {@link Logged} property. + */ + public AnalyticsTable( + AnalyticsTableType tableType, + List columns, + List sortKey, + List primaryKey, + Logged logged) { + super( + toStaging(tableType.getTableName()), + toColumns(columns), + primaryKey, + sortKey, + List.of(), + logged); + this.tableType = tableType; + this.analyticsTableColumns = columns; + } + /** * Constructor. Sets the name to represent a staging table. * @@ -114,6 +142,28 @@ public AnalyticsTable( this.program = program; } + /** + * Constructor. Sets the name to represent a staging table. Use when the table requires a real + * primary key (e.g. Doris continuous analytics). + * + * @param tableType the {@link AnalyticsTableType}. + * @param columns the list of {@link Column}. + * @param primaryKey the primary key. + * @param logged the {@link Logged} property. + * @param program the {@link Program}. + */ + public AnalyticsTable( + AnalyticsTableType tableType, + List columns, + List primaryKey, + Logged logged, + Program program) { + super(toStaging(getTableName(tableType, program)), toColumns(columns), primaryKey, logged); + this.tableType = tableType; + this.analyticsTableColumns = columns; + this.program = program; + } + /** * Constructor. Sets the name to represent a staging table. * diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/model/AnalyticsTablePartition.java b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/model/AnalyticsTablePartition.java index 0f210ad7d618..ce8b2e9efef6 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/model/AnalyticsTablePartition.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/main/java/org/hisp/dhis/analytics/table/model/AnalyticsTablePartition.java @@ -102,6 +102,33 @@ public AnalyticsTablePartition(AnalyticsTable table) { this.endDate = null; } + /** + * Constructor. Sets the name to represent the master table's staging table without a + * partition-specific suffix, for engines where a single physical table serves every logical + * partition (native/declarative partitioning) - while still carrying the given year and date + * range, so callers can determine {@link #isLatestPartition()} and build a correctly-scoped + * populate window against a table name that actually exists. + * + * @param masterTable the master {@link AnalyticsTable} of this partition. + * @param year the year which represents this partition. + * @param startDate the start date of data for this partition. + * @param endDate the end date of data for this partition. + */ + public AnalyticsTablePartition( + AnalyticsTable masterTable, Integer year, Date startDate, Date endDate) { + super( + toStaging(getTableName(masterTable.getMainName(), null)), + List.of(), + List.of(), + List.of(), + masterTable.getLogged(), + masterTable); + this.masterTable = masterTable; + this.year = year; + this.startDate = startDate; + this.endDate = endDate; + } + // ------------------------------------------------------------------------- // Static methods // ------------------------------------------------------------------------- diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/AnalyticsTableServiceTest.java b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/AnalyticsTableServiceTest.java index aa0260015153..34b30ef22969 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/AnalyticsTableServiceTest.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/AnalyticsTableServiceTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2004-2023, University of Oslo + * Copyright (c) 2004-2026, University of Oslo * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -31,17 +31,23 @@ import static org.hisp.dhis.db.model.DataType.DOUBLE; import static org.hisp.dhis.db.model.DataType.TEXT; +import static org.hisp.dhis.test.TestBase.createProgram; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.contains; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.util.Date; import java.util.List; import org.hisp.dhis.analytics.AnalyticsTableManager; import org.hisp.dhis.analytics.AnalyticsTableType; @@ -74,12 +80,6 @@ */ @ExtendWith(MockitoExtension.class) class AnalyticsTableServiceTest { - @Mock private SystemSettingsProvider settingsProvider; - - @Mock private SystemSettings settings; - - @Mock private SqlBuilder sqlBuilder; - @Mock private AnalyticsTableManager tableManager; @Mock private OrganisationUnitService organisationUnitService; @@ -88,6 +88,12 @@ class AnalyticsTableServiceTest { @Mock private ResourceTableService resourceTableService; + @Mock private SystemSettingsProvider settingsProvider; + + @Mock private SystemSettings settings; + + @Mock private SqlBuilder sqlBuilder; + @InjectMocks private DefaultAnalyticsTableService tableService; @Test @@ -119,11 +125,116 @@ void testGetTablePartitions() { new DateTime(2011, 12, 31, 0, 0).toDate()); AnalyticsTable tB = new AnalyticsTable(AnalyticsTableType.ORG_UNIT_TARGET, columns, sortKey, Logged.UNLOGGED); - List partitions = tableService.getTablePartitions(List.of(tA, tB)); + List partitions = + tableService.getTablePartitions(List.of(tA, tB), false); assertEquals(3, partitions.size()); } + @Test + void testGetTablePartitionsUsesRealPartitionForLatestUpdateOnDeclarativePartitioningEngine() { + when(sqlBuilder.supportsDeclarativePartitioning()).thenReturn(true); + when(sqlBuilder.supportsContinuousAnalytics()).thenReturn(true); + + List columns = + List.of( + AnalyticsTableColumn.builder().name("dx").dataType(TEXT).selectExpression("dx").build(), + AnalyticsTableColumn.builder() + .name("value") + .dataType(DOUBLE) + .selectExpression("value") + .build()); + List sortKey = List.of("dx"); + + Date startDate = new DateTime(2026, 7, 31, 12, 48, 20).toDate(); + Date endDate = new DateTime(2026, 7, 31, 12, 55, 0).toDate(); + + AnalyticsTable table = + new AnalyticsTable(AnalyticsTableType.DATA_VALUE, columns, sortKey, Logged.UNLOGGED); + table.addTablePartition( + List.of(), AnalyticsTablePartition.LATEST_PARTITION, startDate, endDate); + + List partitions = + tableService.getTablePartitions(List.of(table), true); + + assertEquals(1, partitions.size()); + AnalyticsTablePartition partition = partitions.get(0); + assertTrue(partition.isLatestPartition()); + assertEquals(startDate, partition.getStartDate()); + assertEquals(endDate, partition.getEndDate()); + assertEquals( + table.getName(), + partition.getName(), + "Populate/create target must be the one physical table this engine actually builds," + + " not a year-suffixed name nothing ever creates"); + } + + @Test + void testGetTablePartitionsUsesFakePartitionForLatestUpdateWithoutContinuousAnalytics() { + // e.g. ClickHouse: with no step removing updated and deleted rows, a continuous update + // replaces the whole main table, so the staging table must hold all data rather than only + // the rows changed within the latest partition's date range. + when(sqlBuilder.supportsDeclarativePartitioning()).thenReturn(true); + when(sqlBuilder.supportsContinuousAnalytics()).thenReturn(false); + + List columns = + List.of( + AnalyticsTableColumn.builder() + .name("dx") + .dataType(TEXT) + .selectExpression("dx") + .build()); + + AnalyticsTable table = + new AnalyticsTable(AnalyticsTableType.DATA_VALUE, columns, List.of("dx"), Logged.UNLOGGED); + table.addTablePartition( + List.of(), + AnalyticsTablePartition.LATEST_PARTITION, + new DateTime(2026, 7, 31, 12, 48, 20).toDate(), + new DateTime(2026, 7, 31, 12, 55, 0).toDate()); + + List partitions = + tableService.getTablePartitions(List.of(table), true); + + assertEquals(1, partitions.size()); + assertFalse(partitions.get(0).isLatestPartition()); + assertNull(partitions.get(0).getStartDate()); + } + + @Test + void testGetTablePartitionsUsesFakePartitionForRegularUpdateOnDeclarativePartitioningEngine() { + when(sqlBuilder.supportsDeclarativePartitioning()).thenReturn(true); + + List columns = + List.of( + AnalyticsTableColumn.builder().name("dx").dataType(TEXT).selectExpression("dx").build(), + AnalyticsTableColumn.builder() + .name("value") + .dataType(DOUBLE) + .selectExpression("value") + .build()); + List sortKey = List.of("dx"); + + AnalyticsTable table = + new AnalyticsTable(AnalyticsTableType.DATA_VALUE, columns, sortKey, Logged.UNLOGGED); + table.addTablePartition( + List.of(), + 2010, + new DateTime(2010, 1, 1, 0, 0).toDate(), + new DateTime(2010, 12, 31, 0, 0).toDate()); + table.addTablePartition( + List.of(), + 2011, + new DateTime(2011, 1, 1, 0, 0).toDate(), + new DateTime(2011, 12, 31, 0, 0).toDate()); + + List partitions = + tableService.getTablePartitions(List.of(table), false); + + assertEquals(1, partitions.size()); + assertFalse(partitions.get(0).isLatestPartition()); + } + @Test void testGetParallelJobsA() { when(settingsProvider.getCurrentSettings()).thenReturn(settings); @@ -141,6 +252,313 @@ void testGetParallelJobsB() { assertEquals(8, tableService.getParallelJobs()); } + @Test + void testRemoveUpdatedDataSkippedWhenContinuousAnalyticsNotSupported() { + AnalyticsTable table = latestPartitionTableFixture(); + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .startTime(new DateTime(2020, 3, 1, 10, 0).toDate()) + .build() + .withLatestPartition(); + + when(settingsProvider.getCurrentSettings()).thenReturn(settings); + when(tableManager.getAnalyticsTableType()).thenReturn(AnalyticsTableType.DATA_VALUE); + when(tableManager.validState()).thenReturn(true); + when(tableManager.getAnalyticsTables(params)).thenReturn(List.of(table)); + when(sqlBuilder.supportsDeclarativePartitioning()).thenReturn(false); + when(sqlBuilder.requiresIndexesForAnalytics()).thenReturn(false); + when(sqlBuilder.supportsAnalyze()).thenReturn(false); + when(sqlBuilder.supportsContinuousAnalytics()).thenReturn(false); + + tableService.create(params, JobProgress.noop()); + + verify(tableManager, never()).removeUpdatedData(anyList()); + } + + @Test + void testRemoveUpdatedDataRunWhenContinuousAnalyticsSupported() { + AnalyticsTable table = latestPartitionTableFixture(); + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .startTime(new DateTime(2020, 3, 1, 10, 0).toDate()) + .build() + .withLatestPartition(); + + when(settingsProvider.getCurrentSettings()).thenReturn(settings); + when(tableManager.getAnalyticsTableType()).thenReturn(AnalyticsTableType.DATA_VALUE); + when(tableManager.validState()).thenReturn(true); + when(tableManager.getAnalyticsTables(params)).thenReturn(List.of(table)); + when(sqlBuilder.supportsDeclarativePartitioning()).thenReturn(false); + when(sqlBuilder.requiresIndexesForAnalytics()).thenReturn(false); + when(sqlBuilder.supportsAnalyze()).thenReturn(false); + when(sqlBuilder.supportsContinuousAnalytics()).thenReturn(true); + when(tableManager.isReadyForContinuousUpdate(List.of(table))).thenReturn(true); + + assertTrue(tableService.create(params, JobProgress.noop())); + + verify(tableManager).removeUpdatedData(List.of(table)); + } + + @Test + void testRemoveUpdatedDataNotRunWhenNotReadyForContinuousUpdate() { + AnalyticsTable table = latestPartitionTableFixture(); + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .startTime(new DateTime(2020, 3, 1, 10, 0).toDate()) + .build() + .withLatestPartition(); + + when(settingsProvider.getCurrentSettings()).thenReturn(settings); + when(tableManager.getAnalyticsTableType()).thenReturn(AnalyticsTableType.DATA_VALUE); + when(tableManager.validState()).thenReturn(true); + when(tableManager.getAnalyticsTables(params)).thenReturn(List.of(table)); + when(sqlBuilder.supportsContinuousAnalytics()).thenReturn(true); + when(tableManager.isReadyForContinuousUpdate(List.of(table))).thenReturn(false); + + JobProgress progress = spy(JobProgress.noop()); + + // e.g. a main table predating unique-key analytics tables on Doris: the delete step must + // never even be attempted against it, not just have its failure tolerated. + assertFalse(tableService.create(params, progress)); + + verify(tableManager, never()).removeUpdatedData(anyList()); + verify(tableManager, never()).swapTable(eq(params), any(AnalyticsTable.class)); + // Checked before any staging work, so a table type that is not ready does not populate its + // staging tables on every scheduled run only to throw them away. + verify(tableManager, never()).preCreateTables(any()); + verify(tableManager, never()).createTable(any()); + // The reason must reach the job log, not only the server log. + verify(progress).failedStage(contains("Run a full analytics table rebuild")); + } + + @Test + void testSwapTableNotRunWhenRemoveUpdatedDataFailsOnUniqueKeyDatabase() { + AnalyticsTable table = latestPartitionTableFixture(); + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .startTime(new DateTime(2020, 3, 1, 10, 0).toDate()) + .build() + .withLatestPartition(); + + when(settingsProvider.getCurrentSettings()).thenReturn(settings); + when(tableManager.getAnalyticsTableType()).thenReturn(AnalyticsTableType.DATA_VALUE); + when(tableManager.validState()).thenReturn(true); + when(tableManager.getAnalyticsTables(params)).thenReturn(List.of(table)); + when(sqlBuilder.supportsDeclarativePartitioning()).thenReturn(false); + when(sqlBuilder.requiresIndexesForAnalytics()).thenReturn(false); + when(sqlBuilder.supportsAnalyze()).thenReturn(false); + when(sqlBuilder.supportsContinuousAnalytics()).thenReturn(true); + when(tableManager.isReadyForContinuousUpdate(List.of(table))).thenReturn(true); + when(tableManager.mainTableExists(table)).thenReturn(true); + when(sqlBuilder.requiresUniqueKeyAnalyticsTables()).thenReturn(true); + doThrow(new IllegalStateException("Delete failed")) + .when(tableManager) + .removeUpdatedData(anyList()); + + // The table's staged data must not be swapped into a main table whose stale/deleted rows + // could not be removed. The update is reported as incomplete, so the window is not advanced. + assertFalse(tableService.create(params, JobProgress.noop())); + + verify(tableManager, never()).swapTable(eq(params), any(AnalyticsTable.class)); + } + + @Test + void testOtherTablesPublishedWhenOneTableDeleteFailsOnUniqueKeyDatabase() { + AnalyticsTable failing = latestPartitionEventTableFixture('A'); + AnalyticsTable working = latestPartitionEventTableFixture('B'); + List tables = List.of(failing, working); + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .startTime(new DateTime(2020, 3, 1, 10, 0).toDate()) + .build() + .withLatestPartition(); + + stubContinuousUpdate(params, tables); + when(tableManager.mainTableExists(failing)).thenReturn(true); + when(tableManager.mainTableExists(working)).thenReturn(true); + doThrow(new IllegalStateException("Delete failed")) + .when(tableManager) + .removeUpdatedData(List.of(failing)); + + JobProgress progress = spy(JobProgress.noop()); + + // The table whose delete failed is not published, the others are. The update is reported as + // incomplete, so the window is not advanced and the next run covers the failed table again. + assertFalse(tableService.create(params, progress)); + + verify(tableManager).removeUpdatedData(List.of(working)); + verify(tableManager).swapTable(params, working); + verify(tableManager, never()).swapTable(params, failing); + verify(progress).failedStage(contains(failing.getMainName())); + } + + @Test + void testTablesSwappedWhenRemoveUpdatedDataFailsWithoutUniqueKeys() { + AnalyticsTable table = latestPartitionTableFixture(); + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .startTime(new DateTime(2020, 3, 1, 10, 0).toDate()) + .build() + .withLatestPartition(); + + when(settingsProvider.getCurrentSettings()).thenReturn(settings); + when(tableManager.getAnalyticsTableType()).thenReturn(AnalyticsTableType.DATA_VALUE); + when(tableManager.validState()).thenReturn(true); + when(tableManager.getAnalyticsTables(params)).thenReturn(List.of(table)); + when(sqlBuilder.supportsDeclarativePartitioning()).thenReturn(false); + when(sqlBuilder.requiresIndexesForAnalytics()).thenReturn(false); + when(sqlBuilder.supportsAnalyze()).thenReturn(false); + when(sqlBuilder.supportsContinuousAnalytics()).thenReturn(true); + when(sqlBuilder.requiresUniqueKeyAnalyticsTables()).thenReturn(false); + when(tableManager.isReadyForContinuousUpdate(List.of(table))).thenReturn(true); + doThrow(new IllegalStateException("Delete failed")) + .when(tableManager) + .removeUpdatedData(anyList()); + + // Postgres: unchanged from before continuous updates on unique-key databases, the tables are + // swapped in after the delete whatever its outcome. + assertTrue(tableService.create(params, JobProgress.noop())); + + verify(tableManager).removeUpdatedData(List.of(table)); + verify(tableManager).swapTable(params, table); + } + + @Test + void testCreateReturnsTrueWhenThereIsNoSourceData() { + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder().build().withLatestPartition(); + + when(settingsProvider.getCurrentSettings()).thenReturn(settings); + when(tableManager.getAnalyticsTableType()).thenReturn(AnalyticsTableType.DATA_VALUE); + when(tableManager.validState()).thenReturn(false); + + // Nothing to publish, so the update is complete and must be recorded as such. + assertTrue(tableService.create(params, JobProgress.noop())); + + verify(tableManager, never()).getAnalyticsTables(any()); + } + + @Test + void testCreateReturnsTrueWhenThereIsNothingToUpdate() { + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder().build().withLatestPartition(); + + when(settingsProvider.getCurrentSettings()).thenReturn(settings); + when(tableManager.getAnalyticsTableType()).thenReturn(AnalyticsTableType.DATA_VALUE); + when(tableManager.validState()).thenReturn(true); + when(tableManager.getAnalyticsTables(params)).thenReturn(List.of()); + + assertTrue(tableService.create(params, JobProgress.noop())); + } + + @Test + void testCreateReturnsFalseWhenCancelled() { + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder().build().withLatestPartition(); + + when(settingsProvider.getCurrentSettings()).thenReturn(settings); + when(tableManager.getAnalyticsTableType()).thenReturn(AnalyticsTableType.DATA_VALUE); + when(tableManager.validState()).thenReturn(true); + JobProgress progress = spy(JobProgress.noop()); + when(progress.isCancelled()).thenReturn(true); + + assertFalse(tableService.create(params, progress)); + + verify(tableManager, never()).getAnalyticsTables(any()); + } + + @Test + void testRemoveUpdatedDataSkipsTableWithoutMainTable() { + // e.g. the event table of a program created after the last full rebuild + AnalyticsTable existing = latestPartitionEventTableFixture('A'); + AnalyticsTable created = latestPartitionEventTableFixture('C'); + List tables = List.of(existing, created); + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .startTime(new DateTime(2020, 3, 1, 10, 0).toDate()) + .build() + .withLatestPartition(); + + stubContinuousUpdate(params, tables); + when(tableManager.mainTableExists(existing)).thenReturn(true); + when(tableManager.mainTableExists(created)).thenReturn(false); + + assertTrue(tableService.create(params, JobProgress.noop())); + + verify(tableManager).removeUpdatedData(List.of(existing)); + verify(tableManager).swapTable(params, existing); + verify(tableManager).swapTable(params, created); + } + + @Test + void testRemoveUpdatedDataNotRunWhenNoMainTableExists() { + AnalyticsTable created = latestPartitionEventTableFixture('C'); + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .startTime(new DateTime(2020, 3, 1, 10, 0).toDate()) + .build() + .withLatestPartition(); + + stubContinuousUpdate(params, List.of(created)); + when(tableManager.mainTableExists(created)).thenReturn(false); + + assertTrue(tableService.create(params, JobProgress.noop())); + + verify(tableManager, never()).removeUpdatedData(anyList()); + verify(tableManager).swapTable(params, created); + } + + private void stubContinuousUpdate( + AnalyticsTableUpdateParams params, List tables) { + when(settingsProvider.getCurrentSettings()).thenReturn(settings); + when(tableManager.getAnalyticsTableType()).thenReturn(AnalyticsTableType.EVENT); + when(tableManager.validState()).thenReturn(true); + when(tableManager.getAnalyticsTables(params)).thenReturn(tables); + when(sqlBuilder.supportsDeclarativePartitioning()).thenReturn(false); + when(sqlBuilder.requiresIndexesForAnalytics()).thenReturn(false); + when(sqlBuilder.supportsAnalyze()).thenReturn(false); + when(sqlBuilder.supportsContinuousAnalytics()).thenReturn(true); + when(sqlBuilder.requiresUniqueKeyAnalyticsTables()).thenReturn(true); + when(tableManager.isReadyForContinuousUpdate(tables)).thenReturn(true); + } + + private AnalyticsTable latestPartitionEventTableFixture(char programCharacter) { + List columns = + List.of( + AnalyticsTableColumn.builder() + .name("event") + .dataType(TEXT) + .selectExpression("event") + .build()); + AnalyticsTable table = + new AnalyticsTable( + AnalyticsTableType.EVENT, columns, Logged.UNLOGGED, createProgram(programCharacter)); + table.addTablePartition( + List.of(), + AnalyticsTablePartition.LATEST_PARTITION, + new DateTime(2020, 1, 1, 0, 0).toDate(), + new DateTime(2020, 3, 1, 10, 0).toDate()); + return table; + } + + private AnalyticsTable latestPartitionTableFixture() { + List columns = + List.of( + AnalyticsTableColumn.builder() + .name("dx") + .dataType(TEXT) + .selectExpression("dx") + .build()); + AnalyticsTable table = + new AnalyticsTable(AnalyticsTableType.DATA_VALUE, columns, List.of(), Logged.UNLOGGED); + table.addTablePartition( + List.of(), + AnalyticsTablePartition.LATEST_PARTITION, + new DateTime(2020, 1, 1, 0, 0).toDate(), + new DateTime(2020, 3, 1, 10, 0).toDate()); + return table; + } + @Test void testRemovesLatestPartitionOverlapForLastYearsUpdate() { AnalyticsTableUpdateParams params = diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/DefaultAnalyticsTableGeneratorTest.java b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/DefaultAnalyticsTableGeneratorTest.java index 4904c3620be9..dc76e1fe6b28 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/DefaultAnalyticsTableGeneratorTest.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/DefaultAnalyticsTableGeneratorTest.java @@ -34,6 +34,7 @@ import static org.hisp.dhis.analytics.AnalyticsTableType.EVENT; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -81,6 +82,8 @@ class DefaultAnalyticsTableGeneratorTest { private void setUp() { when(dataValueTableService.getAnalyticsTableType()).thenReturn(DATA_VALUE); when(eventTableService.getAnalyticsTableType()).thenReturn(EVENT); + lenient().when(dataValueTableService.create(any(), any())).thenReturn(true); + lenient().when(eventTableService.create(any(), any())).thenReturn(true); when(settingsService.getCurrentSettings()).thenReturn(SystemSettings.of(Map.of())); generator = new DefaultAnalyticsTableGenerator( @@ -135,9 +138,31 @@ void skippedTypeGetsNoPerTypeFullUpdateKey_processedTypeDoes() { .put(SystemSettings.keyLastSuccessfulAnalyticsTablesUpdate(DATA_VALUE), startTime); } + @Test + void abortedTypeGetsNoPerTypeLatestPartitionKey_publishedTypeDoes() { + setUp(); + // e.g. the continuous update readiness check or removeUpdatedData() failed for EVENT + when(eventTableService.create(any(), any())).thenReturn(false); + Date startTime = new Date(); + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .skipResourceTables(true) + .startTime(startTime) + .build() + .withLatestPartition(); + + generator.generateAnalyticsTables(params, JobProgress.noop()); + + verify(settingsService, never()) + .put(eq(SystemSettings.keyLastSuccessfulLatestAnalyticsPartitionUpdate(EVENT)), any()); + verify(settingsService) + .put(SystemSettings.keyLastSuccessfulLatestAnalyticsPartitionUpdate(DATA_VALUE), startTime); + } + @Test void typeWithoutLatestPartitionSupportGetsNoPerTypeKeyEvenWhenProcessed() { when(dataValueTableService.getAnalyticsTableType()).thenReturn(ENROLLMENT); + when(dataValueTableService.create(any(), any())).thenReturn(true); when(settingsService.getCurrentSettings()).thenReturn(SystemSettings.of(Map.of())); generator = new DefaultAnalyticsTableGenerator( diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManagerClickHouseTest.java b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManagerClickHouseTest.java new file mode 100644 index 000000000000..acefc85a1a12 --- /dev/null +++ b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManagerClickHouseTest.java @@ -0,0 +1,180 @@ +/* + * Copyright (c) 2004-2026, University of Oslo + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * 1. Redistributions of source code must retain the above copyright notice, this + * list of conditions and the following disclaimer. + * + * 2. Redistributions in binary form must reproduce the above copyright notice, + * this list of conditions and the following disclaimer in the documentation + * and/or other materials provided with the distribution. + * + * 3. Neither the name of the copyright holder nor the names of its contributors + * may be used to endorse or promote products derived from this software without + * specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND + * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED + * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE + * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR + * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES + * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; + * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON + * ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS + * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package org.hisp.dhis.analytics.table; + +import static org.hisp.dhis.db.model.DataType.TEXT; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.when; + +import java.util.Date; +import java.util.List; +import java.util.Map; +import org.hisp.dhis.analytics.AnalyticsTableHookService; +import org.hisp.dhis.analytics.AnalyticsTableType; +import org.hisp.dhis.analytics.AnalyticsTableUpdateParams; +import org.hisp.dhis.analytics.partition.PartitionManager; +import org.hisp.dhis.analytics.table.model.AnalyticsTable; +import org.hisp.dhis.analytics.table.model.AnalyticsTableColumn; +import org.hisp.dhis.analytics.table.setting.AnalyticsTableSettings; +import org.hisp.dhis.category.CategoryService; +import org.hisp.dhis.common.IdentifiableObjectManager; +import org.hisp.dhis.configuration.ConfigurationService; +import org.hisp.dhis.dataapproval.DataApprovalLevelService; +import org.hisp.dhis.dataelement.DataElementService; +import org.hisp.dhis.db.model.Logged; +import org.hisp.dhis.db.sql.ClickHouseSqlBuilder; +import org.hisp.dhis.db.sql.SqlBuilder; +import org.hisp.dhis.organisationunit.OrganisationUnitService; +import org.hisp.dhis.period.PeriodDataProvider; +import org.hisp.dhis.resourcetable.ResourceTableService; +import org.hisp.dhis.setting.SystemSettings; +import org.hisp.dhis.setting.SystemSettingsProvider; +import org.joda.time.DateTime; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.jdbc.core.JdbcTemplate; + +/** + * Covers {@code swapTable()} for databases with declarative partitioning that do NOT support + * continuous analytics (e.g. ClickHouse): with no purge step available, staged data can not be + * merged into an existing main table. The staging table holds all data on such engines, so both a + * continuous and a bounded-years update replace the whole main table, as a full rebuild would. + */ +@ExtendWith(MockitoExtension.class) +class JdbcAnalyticsTableManagerClickHouseTest { + @Mock private IdentifiableObjectManager idObjectManager; + + @Mock private OrganisationUnitService organisationUnitService; + + @Mock private CategoryService categoryService; + + @Mock private SystemSettingsProvider settingsProvider; + + @Mock private SystemSettings settings; + + @Mock private DataApprovalLevelService dataApprovalLevelService; + + @Mock private ResourceTableService resourceTableService; + + @Mock private AnalyticsTableHookService analyticsTableHookService; + + @Mock private PartitionManager partitionManager; + + @Mock private JdbcTemplate jdbcTemplate; + + @Mock private AnalyticsTableSettings analyticsTableSettings; + + @Mock private PeriodDataProvider periodDataProvider; + + @Mock private ConfigurationService configurationService; + + @Mock private DataElementService dataElementService; + + @Spy private SqlBuilder sqlBuilder = new ClickHouseSqlBuilder("dhis2"); + + @InjectMocks private JdbcAnalyticsTableManager subject; + + @BeforeEach + void setUp() { + // shared by the tests that build tables, not needed by every test + lenient().when(settingsProvider.getCurrentSettings()).thenReturn(settings); + lenient().when(settings.getLastSuccessfulResourceTablesUpdate()).thenReturn(new Date(0L)); + } + + @Test + void testSwapTableReplacesMainTableForContinuousUpdateWhenNotSupported() { + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .startTime(new DateTime(2020, 3, 1, 10, 0).toDate()) + .build() + .withLatestPartition(); + + AnalyticsTable table = dataValueTableFixture(); + + // the main table exists; the ClickHouse path replaces it without having to check + lenient() + .when(jdbcTemplate.queryForList(sqlBuilder.tableExists(table.getMainName()))) + .thenReturn(List.of(Map.of("table_name", "analytics"))); + + subject.swapTable(params, table); + + assertReplacedMainTable(); + } + + @Test + void testSwapTableReplacesMainTableForBoundedYearsUpdateOnceMainTableExists() { + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .lastYears(1) + .startTime(new DateTime(2020, 3, 1, 10, 0).toDate()) + .build(); + + AnalyticsTable table = dataValueTableFixture(); + + when(jdbcTemplate.queryForList(sqlBuilder.tableExists(table.getMainName()))) + .thenReturn(List.of(Map.of("table_name", "analytics"))); + + subject.swapTable(params, table); + + assertReplacedMainTable(); + } + + private void assertReplacedMainTable() { + ArgumentCaptor sql = ArgumentCaptor.forClass(String.class); + Mockito.verify(jdbcTemplate, Mockito.atLeastOnce()).execute(sql.capture()); + + List statements = sql.getAllValues(); + assertTrue( + statements.contains("rename table \"analytics_temp\" to \"analytics\";"), + () -> "Expected the staging table to replace the main table, got: " + statements); + assertTrue( + statements.stream().noneMatch(s -> s.startsWith("insert into")), + () -> "Must not merge into the main table without a purge step, got: " + statements); + } + + private AnalyticsTable dataValueTableFixture() { + List columns = + List.of( + AnalyticsTableColumn.builder() + .name("dx") + .dataType(TEXT) + .selectExpression("dx") + .build()); + return new AnalyticsTable(AnalyticsTableType.DATA_VALUE, columns, List.of(), Logged.UNLOGGED); + } +} diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManagerDorisTest.java b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManagerDorisTest.java new file mode 100644 index 000000000000..1f212e030076 --- /dev/null +++ b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManagerDorisTest.java @@ -0,0 +1,459 @@ +/* + * Copyright (c) 2004-2026, University of Oslo + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * 1. Redistributions of source code must retain the above copyright notice, this + * list of conditions and the following disclaimer. + * + * 2. Redistributions in binary form must reproduce the above copyright notice, + * this list of conditions and the following disclaimer in the documentation + * and/or other materials provided with the distribution. + * + * 3. Neither the name of the copyright holder nor the names of its contributors + * may be used to endorse or promote products derived from this software without + * specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND + * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED + * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE + * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR + * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES + * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; + * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON + * ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS + * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package org.hisp.dhis.analytics.table; + +import static java.util.stream.Collectors.toUnmodifiableSet; +import static org.hisp.dhis.db.model.DataType.TEXT; +import static org.hisp.dhis.period.PeriodType.PERIOD_TYPES; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.when; + +import java.util.ArrayList; +import java.util.Date; +import java.util.List; +import java.util.Map; +import org.hisp.dhis.analytics.AnalyticsTableHookService; +import org.hisp.dhis.analytics.AnalyticsTableType; +import org.hisp.dhis.analytics.AnalyticsTableUpdateParams; +import org.hisp.dhis.analytics.partition.PartitionManager; +import org.hisp.dhis.analytics.table.model.AnalyticsTable; +import org.hisp.dhis.analytics.table.model.AnalyticsTableColumn; +import org.hisp.dhis.analytics.table.model.AnalyticsTablePartition; +import org.hisp.dhis.analytics.table.setting.AnalyticsTableSettings; +import org.hisp.dhis.category.CategoryService; +import org.hisp.dhis.common.IdentifiableObjectManager; +import org.hisp.dhis.configuration.Configuration; +import org.hisp.dhis.configuration.ConfigurationService; +import org.hisp.dhis.dataapproval.DataApprovalLevelService; +import org.hisp.dhis.dataelement.DataElementService; +import org.hisp.dhis.db.model.Logged; +import org.hisp.dhis.db.sql.DorisSqlBuilder; +import org.hisp.dhis.db.sql.SqlBuilder; +import org.hisp.dhis.organisationunit.OrganisationUnitService; +import org.hisp.dhis.period.PeriodDataProvider; +import org.hisp.dhis.resourcetable.ResourceTableService; +import org.hisp.dhis.setting.SystemSettings; +import org.hisp.dhis.setting.SystemSettingsProvider; +import org.joda.time.DateTime; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.jdbc.core.JdbcTemplate; + +@ExtendWith(MockitoExtension.class) +class JdbcAnalyticsTableManagerDorisTest { + @Mock private IdentifiableObjectManager idObjectManager; + + @Mock private OrganisationUnitService organisationUnitService; + + @Mock private CategoryService categoryService; + + @Mock private SystemSettingsProvider settingsProvider; + + @Mock private SystemSettings settings; + + @Mock private DataApprovalLevelService dataApprovalLevelService; + + @Mock private ResourceTableService resourceTableService; + + @Mock private AnalyticsTableHookService analyticsTableHookService; + + @Mock private PartitionManager partitionManager; + + @Mock private JdbcTemplate jdbcTemplate; + + @Mock private AnalyticsTableSettings analyticsTableSettings; + + @Mock private PeriodDataProvider periodDataProvider; + + @Mock private ConfigurationService configurationService; + + @Mock private Configuration configuration; + + @Mock private DataElementService dataElementService; + + @Spy private SqlBuilder sqlBuilder = new DorisSqlBuilder("dhis2", "driver"); + + @InjectMocks private JdbcAnalyticsTableManager subject; + + @BeforeEach + void setUp() { + // shared by the tests that build tables, not needed by every test + lenient().when(settingsProvider.getCurrentSettings()).thenReturn(settings); + lenient().when(settings.getLastSuccessfulResourceTablesUpdate()).thenReturn(new Date(0L)); + } + + @Test + void testGetRegularAnalyticsTableHasUniqueKeyOnId() { + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .startTime(new DateTime(2020, 1, 1, 0, 0).toDate()) + .build(); + + when(jdbcTemplate.queryForList( + org.mockito.Mockito.anyString(), org.mockito.Mockito.eq(Integer.class))) + .thenReturn(List.of(2020)); + when(configurationService.getConfiguration()).thenReturn(configuration); + when(configuration.getDataOutputPeriodTypes()) + .thenReturn(PERIOD_TYPES.stream().collect(toUnmodifiableSet())); + + List tables = subject.getAnalyticsTables(params); + + assertEquals(1, tables.size()); + AnalyticsTable table = tables.get(0); + assertTrue(table.hasPrimaryKey()); + assertEquals(List.of("id", "year"), table.getPrimaryKey()); + String createTableSql = sqlBuilder.createTable(table); + assertTrue(createTableSql.contains("unique key (`id`,`year`)")); + // Doris rejects unbounded string/text types as key columns on any key model (errCode = 2, + // "String Type should not be used in key column"). The id column (a concat(...) of several + // dimension UIDs) must be given a bounded type on Doris, unlike Postgres/ClickHouse where it + // stays TEXT. + assertTrue( + createTableSql.contains("`id` varchar(1020)"), + () -> "Expected id column to use a bounded varchar type, got: " + createTableSql); + assertFalse( + createTableSql.contains("`id` string"), + () -> + "id column must not use the unbounded string type as a key column, got: " + + createTableSql); + } + + @Test + @DisplayName( + "Doris latest-partition window starts at the last continuous update, not the last full" + + " rebuild, since Doris merges into the persistent main table rather than replacing a" + + " partition wholesale") + void testGetLatestAnalyticsTableUsesLastAnyTableUpdate() { + Date lastFullTableUpdate = new DateTime(2019, 3, 1, 2, 0).toDate(); + Date lastLatestPartitionUpdate = new DateTime(2019, 3, 1, 9, 0).toDate(); + Date startTime = new DateTime(2019, 3, 1, 10, 0).toDate(); + + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder().startTime(startTime).build().withLatestPartition(); + + List> queryResp = new ArrayList<>(); + queryResp.add(Map.of("dataelementid", 1)); + + when(settings.getLastSuccessfulAnalyticsTablesUpdate()).thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulAnalyticsTablesUpdate(AnalyticsTableType.DATA_VALUE)) + .thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulLatestAnalyticsPartitionUpdate(AnalyticsTableType.DATA_VALUE)) + .thenReturn(lastLatestPartitionUpdate); + when(analyticsTableSettings.getTableLogged()).thenReturn(Logged.UNLOGGED); + when(jdbcTemplate.queryForList(org.mockito.Mockito.anyString())).thenReturn(queryResp); + when(configurationService.getConfiguration()).thenReturn(configuration); + when(configuration.getDataOutputPeriodTypes()) + .thenReturn(PERIOD_TYPES.stream().collect(toUnmodifiableSet())); + + List tables = subject.getAnalyticsTables(params); + + assertEquals(1, tables.size()); + + AnalyticsTablePartition partition = tables.get(0).getLatestTablePartition(); + + assertTrue(partition.isLatestPartition()); + assertEquals(lastLatestPartitionUpdate, partition.getStartDate()); + assertEquals(startTime, partition.getEndDate()); + } + + @Test + @DisplayName( + "Doris latest-partition window starts at this table type's own last update, not the shared" + + " one, which also advances when this type was skipped or its update was aborted") + void testGetLatestAnalyticsTableUsesPerTypeLastUpdateNotShared() { + Date lastFullTableUpdate = new DateTime(2019, 3, 1, 2, 0).toDate(); + Date lastDataValueUpdate = new DateTime(2019, 3, 1, 5, 0).toDate(); + Date lastSharedUpdate = new DateTime(2019, 3, 1, 9, 0).toDate(); + Date startTime = new DateTime(2019, 3, 1, 10, 0).toDate(); + + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder().startTime(startTime).build().withLatestPartition(); + + List> queryResp = new ArrayList<>(); + queryResp.add(Map.of("dataelementid", 1)); + + when(settings.getLastSuccessfulAnalyticsTablesUpdate()).thenReturn(lastFullTableUpdate); + lenient() + .when(settings.getLastSuccessfulLatestAnalyticsPartitionUpdate()) + .thenReturn(lastSharedUpdate); + when(settings.getLastSuccessfulAnalyticsTablesUpdate(AnalyticsTableType.DATA_VALUE)) + .thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulLatestAnalyticsPartitionUpdate(AnalyticsTableType.DATA_VALUE)) + .thenReturn(lastDataValueUpdate); + when(analyticsTableSettings.getTableLogged()).thenReturn(Logged.UNLOGGED); + when(jdbcTemplate.queryForList(org.mockito.Mockito.anyString())).thenReturn(queryResp); + when(configurationService.getConfiguration()).thenReturn(configuration); + when(configuration.getDataOutputPeriodTypes()) + .thenReturn(PERIOD_TYPES.stream().collect(toUnmodifiableSet())); + + List tables = subject.getAnalyticsTables(params); + + assertEquals(1, tables.size()); + assertEquals(lastDataValueUpdate, tables.get(0).getLatestTablePartition().getStartDate()); + } + + @Test + void testRemoveUpdatedDataMaterializesKeysNatively() { + Date lastFullTableUpdate = new DateTime(2019, 3, 1, 2, 0).toDate(); + Date lastLatestPartitionUpdate = new DateTime(2019, 3, 1, 9, 0).toDate(); + Date startTime = new DateTime(2019, 3, 1, 10, 0).toDate(); + + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder().startTime(startTime).build().withLatestPartition(); + + List> queryResp = new ArrayList<>(); + queryResp.add(Map.of("dataelementid", 1)); + + when(settings.getLastSuccessfulAnalyticsTablesUpdate()).thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulAnalyticsTablesUpdate(AnalyticsTableType.DATA_VALUE)) + .thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulLatestAnalyticsPartitionUpdate(AnalyticsTableType.DATA_VALUE)) + .thenReturn(lastLatestPartitionUpdate); + when(jdbcTemplate.queryForList(org.mockito.Mockito.anyString())).thenReturn(queryResp); + when(configurationService.getConfiguration()).thenReturn(configuration); + when(configuration.getDataOutputPeriodTypes()) + .thenReturn(PERIOD_TYPES.stream().collect(toUnmodifiableSet())); + + List tables = subject.getAnalyticsTables(params); + assertEquals(1, tables.size()); + + subject.removeUpdatedData(tables); + + org.mockito.ArgumentCaptor sql = org.mockito.ArgumentCaptor.forClass(String.class); + org.mockito.Mockito.verify(jdbcTemplate, org.mockito.Mockito.times(5)).execute(sql.capture()); + List statements = sql.getAllValues(); + + String drop1 = statements.get(0); + assertTrue(drop1.toLowerCase().contains("drop table if exists")); + assertTrue(drop1.contains("analytics_delete_keys")); + + String create = statements.get(1); + assertTrue(create.toLowerCase().contains("create table")); + assertTrue(create.contains("analytics_delete_keys")); + assertTrue(create.contains("`dx`")); + assertTrue(create.contains("`co`")); + assertTrue(create.contains("`ao`")); + assertTrue(create.contains("`pe`")); + assertTrue(create.contains("`ou`")); + assertTrue(create.contains("`year`")); + assertFalse(create.contains("varchar(1020)")); + + String insert = statements.get(2); + assertTrue(insert.toLowerCase().contains("insert into")); + assertTrue(insert.contains("analytics_delete_keys")); + assertTrue(insert.contains("datavalue")); + assertTrue(insert.contains("analytics_rs_dataelementstructure")); + assertFalse(insert.contains("concat(")); + assertTrue(insert.contains("dv.lastupdated >=")); + assertTrue(insert.contains("dv.lastupdated <")); + assertTrue( + insert.contains( + "(dx,co,ao,pe,ou,year) select des.dataelementuid, dcs.categoryoptioncombouid, " + + "acs.categoryoptioncombouid, ps.iso, ous.organisationunituid, ps.year"), + () -> "Expected column-list-to-select-list pairing to match exactly, got: " + insert); + + String delete = statements.get(3); + assertTrue(delete.toLowerCase().contains("delete from")); + assertTrue(delete.toLowerCase().contains("using")); + assertTrue(delete.contains("analytics_delete_keys")); + assertTrue(delete.contains("ax.dx=k.dx")); + assertTrue(delete.contains("ax.co=k.co")); + assertTrue(delete.contains("ax.ao=k.ao")); + assertTrue(delete.contains("ax.pe=k.pe")); + assertTrue(delete.contains("ax.ou=k.ou")); + assertTrue(delete.contains("ax.year=k.year")); + assertFalse(delete.contains("datavalue")); + assertFalse(delete.contains("analytics_rs_")); + + String drop2 = statements.get(4); + assertTrue(drop2.toLowerCase().contains("drop table")); + assertTrue(drop2.contains("analytics_delete_keys")); + } + + @Test + void testSwapTableInsertsStagingDataIntoMainTableWhenSkippingMasterTable() { + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .startTime(new DateTime(2020, 3, 1, 10, 0).toDate()) + .build() + .withLatestPartition(); + + List columns = + List.of( + AnalyticsTableColumn.builder() + .name("dx") + .dataType(TEXT) + .selectExpression("dx") + .build()); + AnalyticsTable table = + new AnalyticsTable(AnalyticsTableType.DATA_VALUE, columns, List.of(), Logged.UNLOGGED); + + // Main table already exists, and params.isPartialUpdate() (via withLatestPartition()) plus + // AnalyticsTableType.DATA_VALUE.isLatestPartition()==true together push swapTable() into the + // skipMasterTable branch. + when(jdbcTemplate.queryForList(sqlBuilder.tableExists(table.getMainName()))) + .thenReturn(List.of(Map.of("table_name", "analytics"))); + + subject.swapTable(params, table); + + org.mockito.ArgumentCaptor sql = org.mockito.ArgumentCaptor.forClass(String.class); + org.mockito.Mockito.verify(jdbcTemplate, org.mockito.Mockito.times(2)).execute(sql.capture()); + + List statements = sql.getAllValues(); + assertTrue( + statements.stream() + .anyMatch( + s -> + s.startsWith("insert into `analytics`") && s.contains("from `analytics_temp`")), + () -> "Expected an insert-into-main-from-staging statement, got: " + statements); + // The staging table is a native Doris table, not a federated one, so the FROM reference must + // NOT be qualified with the federated catalog prefix that qualifyTable() would add. + assertTrue( + statements.stream().noneMatch(s -> s.contains("dhis2.public.")), + () -> + "Staging table reference must not carry the federated catalog prefix, got: " + + statements); + } + + @Test + void testSwapTableReplacesMainTableForBoundedYearsUpdateOnceMainTableExists() { + // lastYears(1) (a normal bounded-years rebuild) makes isPartialUpdate()==true but + // isLatestUpdate()==false -- distinct from the "latest partition" (lastYears=0) case the + // insert-into-main branch is scoped to. removeUpdatedData() never runs for this case (it's + // gated on isLatestUpdate()), so merging staging data into main here would leave deleted rows + // behind forever, and dropping the staging data would leave stale values. The staging table + // holds every year on Doris (no year filter on declarative partitioning), so the whole main + // table is replaced instead. + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .lastYears(1) + .startTime(new DateTime(2020, 3, 1, 10, 0).toDate()) + .build(); + + List columns = + List.of( + AnalyticsTableColumn.builder() + .name("dx") + .dataType(TEXT) + .selectExpression("dx") + .build()); + AnalyticsTable table = + new AnalyticsTable(AnalyticsTableType.DATA_VALUE, columns, List.of(), Logged.UNLOGGED); + + // Same preconditions as the merge test: main table already exists, + // params.isPartialUpdate()==true (via lastYears(1)), and DATA_VALUE.isLatestPartition()==true. + when(jdbcTemplate.queryForList(sqlBuilder.tableExists(table.getMainName()))) + .thenReturn(List.of(Map.of("table_name", "analytics"))); + + subject.swapTable(params, table); + + org.mockito.ArgumentCaptor sql = org.mockito.ArgumentCaptor.forClass(String.class); + org.mockito.Mockito.verify(jdbcTemplate, org.mockito.Mockito.atLeastOnce()) + .execute(sql.capture()); + + List statements = sql.getAllValues(); + assertTrue( + statements.stream() + .anyMatch(s -> s.contains("alter table `analytics_temp` rename `analytics`;")), + () -> "Expected the staging table to replace the main table, got: " + statements); + assertTrue( + statements.stream().noneMatch(s -> s.startsWith("insert into")), + () -> "Bounded update must not merge into the main table, got: " + statements); + } + + @Test + void testIsReadyForContinuousUpdateWhenMainTableDoesNotExistYet() { + AnalyticsTable table = dataValueTableFixture(); + + when(jdbcTemplate.queryForList(sqlBuilder.tableExists(table.getMainName()))) + .thenReturn(List.of()); + + assertTrue(subject.isReadyForContinuousUpdate(List.of(table))); + + org.mockito.Mockito.verify(jdbcTemplate, org.mockito.Mockito.never()) + .queryForList(sqlBuilder.showCreateTable(table.getMainName())); + } + + @Test + void testIsReadyForContinuousUpdateWhenMainTableHasUniqueKey() { + AnalyticsTable table = dataValueTableFixture(); + + when(jdbcTemplate.queryForList(sqlBuilder.tableExists(table.getMainName()))) + .thenReturn(List.of(Map.of("table_name", "analytics"))); + when(jdbcTemplate.queryForList(sqlBuilder.showCreateTable(table.getMainName()))) + .thenReturn( + List.of( + Map.of( + "Table", "analytics", + "Create Table", + "CREATE TABLE `analytics` (...) UNIQUE KEY (`id`,`year`) ..."))); + + assertTrue(subject.isReadyForContinuousUpdate(List.of(table))); + } + + @Test + void testIsReadyForContinuousUpdateWhenMainTablePredatesUniqueKeySupport() { + // A main table built before unique-key analytics tables were introduced remains on the + // duplicate key model until a full rebuild recreates it. The delete step a continuous update + // depends on only works on unique-key tables, so it must not be attempted. + AnalyticsTable table = dataValueTableFixture(); + + when(jdbcTemplate.queryForList(sqlBuilder.tableExists(table.getMainName()))) + .thenReturn(List.of(Map.of("table_name", "analytics"))); + when(jdbcTemplate.queryForList(sqlBuilder.showCreateTable(table.getMainName()))) + .thenReturn( + List.of( + Map.of( + "Table", "analytics", + "Create Table", "CREATE TABLE `analytics` (...) DUPLICATE KEY (`id`) ..."))); + + assertFalse(subject.isReadyForContinuousUpdate(List.of(table))); + } + + private AnalyticsTable dataValueTableFixture() { + List columns = + List.of( + AnalyticsTableColumn.builder() + .name("dx") + .dataType(TEXT) + .selectExpression("dx") + .build()); + return new AnalyticsTable(AnalyticsTableType.DATA_VALUE, columns, List.of(), Logged.UNLOGGED); + } +} diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManagerTest.java b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManagerTest.java index 253a0f7f7f93..752585afc24b 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManagerTest.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcAnalyticsTableManagerTest.java @@ -33,6 +33,7 @@ import static org.hisp.dhis.analytics.AnalyticsStringUtils.qualifyVariables; import static org.hisp.dhis.analytics.AnalyticsStringUtils.replaceQualify; import static org.hisp.dhis.db.model.DataType.INTEGER; +import static org.hisp.dhis.db.model.DataType.TEXT; import static org.hisp.dhis.db.model.Logged.LOGGED; import static org.hisp.dhis.db.model.Logged.UNLOGGED; import static org.hisp.dhis.period.PeriodType.PERIOD_TYPES; @@ -66,6 +67,7 @@ import org.hisp.dhis.configuration.Configuration; import org.hisp.dhis.configuration.ConfigurationService; import org.hisp.dhis.dataapproval.DataApprovalLevelService; +import org.hisp.dhis.db.model.Column; import org.hisp.dhis.db.model.Table; import org.hisp.dhis.db.sql.PostgreSqlBuilder; import org.hisp.dhis.db.sql.SqlBuilder; @@ -222,6 +224,35 @@ void testGetRegularAnalyticsTable() { partitionB.getYear().intValue(), new DateTime(partitionB.getStartDate()).getYear()); } + @Test + void testGetRegularAnalyticsTableIdColumnRemainsTextOnPostgres() { + // Doris needs a bounded type for the id key column (see JdbcAnalyticsTableManagerDorisTest), + // but that must not change Postgres behavior, since + // PostgreSqlBuilder.requiresUniqueKeyAnalyticsTables() + // is false and id is not used as a key column there. + Date startTime = new DateTime(2019, 3, 1, 10, 0).toDate(); + List dataYears = List.of(2018, 2019); + + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder().startTime(startTime).build(); + + when(analyticsTableSettings.getTableLogged()).thenReturn(UNLOGGED); + when(jdbcTemplate.queryForList(Mockito.anyString(), ArgumentMatchers.>any())) + .thenReturn(dataYears); + when(configurationService.getConfiguration()).thenReturn(configuration); + when(configuration.getDataOutputPeriodTypes()) + .thenReturn(PERIOD_TYPES.stream().collect(toUnmodifiableSet())); + + List tables = subject.getAnalyticsTables(params); + + assertEquals(1, tables.size()); + AnalyticsTable table = tables.get(0); + Column idColumn = + table.getColumns().stream().filter(c -> "id".equals(c.getName())).findFirst().orElseThrow(); + + assertEquals(TEXT, idColumn.getDataType()); + } + @Test void testGetRegularAnalyticsTableLogged() { Date startTime = new DateTime(2019, 3, 1, 10, 0).toDate(); diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcCompletenessTableManagerDorisTest.java b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcCompletenessTableManagerDorisTest.java new file mode 100644 index 000000000000..d3586ec1ef38 --- /dev/null +++ b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcCompletenessTableManagerDorisTest.java @@ -0,0 +1,189 @@ +/* + * Copyright (c) 2004-2026, University of Oslo + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * 1. Redistributions of source code must retain the above copyright notice, this + * list of conditions and the following disclaimer. + * + * 2. Redistributions in binary form must reproduce the above copyright notice, + * this list of conditions and the following disclaimer in the documentation + * and/or other materials provided with the distribution. + * + * 3. Neither the name of the copyright holder nor the names of its contributors + * may be used to endorse or promote products derived from this software without + * specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND + * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED + * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE + * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR + * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES + * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; + * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON + * ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS + * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package org.hisp.dhis.analytics.table; + +import static java.util.stream.Collectors.toUnmodifiableSet; +import static org.hisp.dhis.period.PeriodType.PERIOD_TYPES; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.anyString; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.ArrayList; +import java.util.Date; +import java.util.List; +import java.util.Map; +import org.hisp.dhis.analytics.AnalyticsTableHookService; +import org.hisp.dhis.analytics.AnalyticsTableType; +import org.hisp.dhis.analytics.AnalyticsTableUpdateParams; +import org.hisp.dhis.analytics.partition.PartitionManager; +import org.hisp.dhis.analytics.table.model.AnalyticsTable; +import org.hisp.dhis.analytics.table.setting.AnalyticsTableSettings; +import org.hisp.dhis.category.CategoryService; +import org.hisp.dhis.common.IdentifiableObjectManager; +import org.hisp.dhis.configuration.Configuration; +import org.hisp.dhis.configuration.ConfigurationService; +import org.hisp.dhis.dataapproval.DataApprovalLevelService; +import org.hisp.dhis.db.sql.DorisSqlBuilder; +import org.hisp.dhis.db.sql.SqlBuilder; +import org.hisp.dhis.organisationunit.OrganisationUnitService; +import org.hisp.dhis.period.PeriodDataProvider; +import org.hisp.dhis.resourcetable.ResourceTableService; +import org.hisp.dhis.setting.SystemSettings; +import org.hisp.dhis.setting.SystemSettingsProvider; +import org.joda.time.DateTime; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.jdbc.core.JdbcTemplate; + +@ExtendWith(MockitoExtension.class) +class JdbcCompletenessTableManagerDorisTest { + @Mock private IdentifiableObjectManager idObjectManager; + + @Mock private OrganisationUnitService organisationUnitService; + + @Mock private CategoryService categoryService; + + @Mock private SystemSettingsProvider settingsProvider; + + @Mock private SystemSettings settings; + + @Mock private DataApprovalLevelService dataApprovalLevelService; + + @Mock private ResourceTableService resourceTableService; + + @Mock private AnalyticsTableHookService analyticsTableHookService; + + @Mock private PartitionManager partitionManager; + + @Mock private JdbcTemplate jdbcTemplate; + + @Mock private AnalyticsTableSettings analyticsTableSettings; + + @Mock private PeriodDataProvider periodDataProvider; + + @Mock private ConfigurationService configurationService; + + @Mock private Configuration configuration; + + @Spy private SqlBuilder sqlBuilder = new DorisSqlBuilder("dhis2", "driver"); + + @InjectMocks private JdbcCompletenessTableManager subject; + + @BeforeEach + void setUp() { + when(settingsProvider.getCurrentSettings()).thenReturn(settings); + when(settings.getLastSuccessfulResourceTablesUpdate()).thenReturn(new Date(0L)); + when(configurationService.getConfiguration()).thenReturn(configuration); + when(configuration.getDataOutputPeriodTypes()) + .thenReturn(PERIOD_TYPES.stream().collect(toUnmodifiableSet())); + } + + @Test + void testGetLatestAnalyticsTableHasUniqueKeyOnId() { + Date lastFullTableUpdate = new DateTime(2019, 3, 1, 2, 0).toDate(); + Date lastLatestPartitionUpdate = new DateTime(2019, 3, 1, 9, 0).toDate(); + Date startTime = new DateTime(2019, 3, 1, 10, 0).toDate(); + + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder().startTime(startTime).build().withLatestPartition(); + + List> queryResp = new ArrayList<>(); + queryResp.add(Map.of("datasetid", 1)); + + when(settings.getLastSuccessfulAnalyticsTablesUpdate()).thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulAnalyticsTablesUpdate(AnalyticsTableType.COMPLETENESS)) + .thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulLatestAnalyticsPartitionUpdate(AnalyticsTableType.COMPLETENESS)) + .thenReturn(lastLatestPartitionUpdate); + when(jdbcTemplate.queryForList(anyString())).thenReturn(queryResp); + + List tables = subject.getAnalyticsTables(params); + + assertEquals(1, tables.size()); + AnalyticsTable table = tables.get(0); + assertTrue(table.hasPrimaryKey()); + assertEquals(List.of("id", "year"), table.getPrimaryKey()); + String createTableSql = sqlBuilder.createTable(table); + assertTrue(createTableSql.contains("unique key (`id`,`year`)")); + // Doris rejects unbounded string/text types as key columns on any key model (errCode = 2, + // "String Type should not be used in key column"). The id column (a concat(...) of several + // dimension UIDs) must be given a bounded type on Doris, unlike Postgres/ClickHouse where it + // stays TEXT. + assertTrue( + createTableSql.contains("`id` varchar(1020)"), + () -> "Expected id column to use a bounded varchar type, got: " + createTableSql); + assertFalse( + createTableSql.contains("`id` string"), + () -> + "id column must not use the unbounded string type as a key column, got: " + + createTableSql); + } + + @Test + void testRemoveUpdatedDataUsesUsingJoinNotSubquery() { + Date lastFullTableUpdate = new DateTime(2019, 3, 1, 2, 0).toDate(); + Date lastLatestPartitionUpdate = new DateTime(2019, 3, 1, 9, 0).toDate(); + Date startTime = new DateTime(2019, 3, 1, 10, 0).toDate(); + + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder().startTime(startTime).build().withLatestPartition(); + + List> queryResp = new ArrayList<>(); + queryResp.add(Map.of("datasetid", 1)); + + when(settings.getLastSuccessfulAnalyticsTablesUpdate()).thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulAnalyticsTablesUpdate(AnalyticsTableType.COMPLETENESS)) + .thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulLatestAnalyticsPartitionUpdate(AnalyticsTableType.COMPLETENESS)) + .thenReturn(lastLatestPartitionUpdate); + when(jdbcTemplate.queryForList(anyString())).thenReturn(queryResp); + + List tables = subject.getAnalyticsTables(params); + assertEquals(1, tables.size()); + + subject.removeUpdatedData(tables); + + ArgumentCaptor sql = ArgumentCaptor.forClass(String.class); + verify(jdbcTemplate).execute(sql.capture()); + + assertTrue(sql.getValue().contains("using")); + assertTrue(sql.getValue().contains("ax.id =")); + assertFalse(sql.getValue().contains("in (")); + } +} diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcEventAnalyticsTableManagerDorisTest.java b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcEventAnalyticsTableManagerDorisTest.java index 99d26c4b8ebe..362c771f7128 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcEventAnalyticsTableManagerDorisTest.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcEventAnalyticsTableManagerDorisTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2004-2022, University of Oslo + * Copyright (c) 2004-2026, University of Oslo * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -49,12 +49,18 @@ import static org.hisp.dhis.test.TestBase.createProgramStage; import static org.hisp.dhis.test.TestBase.createProgramTrackedEntityAttribute; import static org.hisp.dhis.test.TestBase.createTrackedEntityAttribute; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.time.LocalDate; import java.time.ZoneId; +import java.util.ArrayList; import java.util.Date; import java.util.List; +import java.util.Map; import java.util.Set; import org.hisp.dhis.analytics.AggregationType; import org.hisp.dhis.analytics.AnalyticsTableType; @@ -83,15 +89,19 @@ import org.hisp.dhis.program.Program; import org.hisp.dhis.program.ProgramStage; import org.hisp.dhis.program.ProgramTrackedEntityAttribute; +import org.hisp.dhis.program.ProgramType; import org.hisp.dhis.resourcetable.ResourceTableService; import org.hisp.dhis.setting.SystemSettings; import org.hisp.dhis.setting.SystemSettingsProvider; import org.hisp.dhis.trackedentity.TrackedEntityAttribute; import org.joda.time.DateTime; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; +import org.mockito.Mockito; import org.mockito.Spy; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.jdbc.core.JdbcTemplate; @@ -291,4 +301,221 @@ void verifyGetTableWithDataElements() { private String toSelectExpression(String template, String uid) { return String.format(template, uid, uid, uid); } + + @Test + void testRegularAnalyticsTableHasUniqueKeyOnEvent() { + Program program = createProgram('A'); + program.setProgramType(ProgramType.WITH_REGISTRATION); + + when(configurationService.getConfiguration()).thenReturn(configuration); + when(configuration.getDataOutputPeriodTypes()) + .thenReturn(PERIOD_TYPES.stream().collect(toUnmodifiableSet())); + when(idObjectManager.getAllNoAcl(Program.class)).thenReturn(List.of(program)); + when(periodDataProvider.getAvailableYears(DATABASE)).thenReturn(List.of(2018, now().getYear())); + when(jdbcTemplate.queryForList(Mockito.anyString(), Mockito.eq(Integer.class))) + .thenReturn(List.of(2018)); + + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder() + .lastYears(2) + .startTime(START_TIME) + .today(today) + .build(); + + List tables = subject.getAnalyticsTables(params); + + assertThat(tables, hasSize(1)); + AnalyticsTable table = tables.get(0); + assertTrue(table.hasPrimaryKey()); + assertEquals(List.of("event", "year"), table.getPrimaryKey()); + assertTrue(sqlBuilder.createTable(table).contains("unique key (`event`,`year`)")); + } + + @Test + @DisplayName( + "A program without a main table yet gets all its data in a continuous update, since its" + + " staging table becomes the main table; a program with one gets the continuous window") + void testLatestTableOfProgramWithoutMainTableStartsAtEpoch() { + Program existing = createProgram('A'); + Program created = createProgram('C'); + + Date lastFullTableUpdate = new DateTime(2019, 3, 1, 2, 0).toDate(); + Date lastLatestPartitionUpdate = new DateTime(2019, 3, 1, 9, 0).toDate(); + Date startTime = new DateTime(2019, 3, 1, 10, 0).toDate(); + + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder().startTime(startTime).build().withLatestPartition(); + + List> queryResp = new ArrayList<>(); + queryResp.add(Map.of("eventid", 1)); + + when(settings.getLastSuccessfulAnalyticsTablesUpdate()).thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulAnalyticsTablesUpdate(AnalyticsTableType.EVENT)) + .thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulLatestAnalyticsPartitionUpdate(AnalyticsTableType.EVENT)) + .thenReturn(lastLatestPartitionUpdate); + when(jdbcTemplate.queryForList(Mockito.anyString())).thenReturn(queryResp); + when(jdbcTemplate.queryForList( + sqlBuilder.tableExists(AnalyticsTable.getTableName(AnalyticsTableType.EVENT, created)))) + .thenReturn(List.of()); + when(idObjectManager.getAllNoAcl(Program.class)).thenReturn(List.of(existing, created)); + when(configurationService.getConfiguration()).thenReturn(configuration); + when(configuration.getDataOutputPeriodTypes()) + .thenReturn(PERIOD_TYPES.stream().collect(toUnmodifiableSet())); + + List tables = subject.getAnalyticsTables(params); + assertThat(tables, hasSize(2)); + + assertEquals(lastLatestPartitionUpdate, tables.get(0).getLatestTablePartition().getStartDate()); + assertEquals(new Date(0L), tables.get(1).getLatestTablePartition().getStartDate()); + assertEquals(startTime, tables.get(1).getLatestTablePartition().getEndDate()); + } + + @Test + @DisplayName( + "removeUpdatedData materializes event keys natively for a registration program on Doris," + + " instead of a federated join inside DELETE") + void testRemoveUpdatedDataMaterializesKeysNativelyForRegistration() { + Program program = createProgram('A'); + program.setProgramType(ProgramType.WITH_REGISTRATION); + + Date lastFullTableUpdate = new DateTime(2019, 3, 1, 2, 0).toDate(); + Date lastLatestPartitionUpdate = new DateTime(2019, 3, 1, 9, 0).toDate(); + Date startTime = new DateTime(2019, 3, 1, 10, 0).toDate(); + + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder().startTime(startTime).build().withLatestPartition(); + + List> queryResp = new ArrayList<>(); + queryResp.add(Map.of("eventid", 1)); + + when(settings.getLastSuccessfulAnalyticsTablesUpdate()).thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulAnalyticsTablesUpdate(AnalyticsTableType.EVENT)) + .thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulLatestAnalyticsPartitionUpdate(AnalyticsTableType.EVENT)) + .thenReturn(lastLatestPartitionUpdate); + when(jdbcTemplate.queryForList(Mockito.anyString())).thenReturn(queryResp); + when(idObjectManager.getAllNoAcl(Program.class)).thenReturn(List.of(program)); + when(configurationService.getConfiguration()).thenReturn(configuration); + when(configuration.getDataOutputPeriodTypes()) + .thenReturn(PERIOD_TYPES.stream().collect(toUnmodifiableSet())); + + List tables = subject.getAnalyticsTables(params); + assertThat(tables, hasSize(1)); + + subject.removeUpdatedData(tables); + + ArgumentCaptor sql = ArgumentCaptor.forClass(String.class); + verify(jdbcTemplate, Mockito.times(5)).execute(sql.capture()); + List statements = sql.getAllValues(); + + String drop1 = statements.get(0); + assertTrue(drop1.toLowerCase().contains("drop table if exists")); + assertTrue(drop1.contains("analytics_event_delete_keys")); + + String create = statements.get(1); + assertTrue(create.toLowerCase().contains("create table")); + assertTrue(create.contains("analytics_event_delete_keys")); + assertTrue(create.contains("`event`")); + assertFalse(create.contains("`year`")); + + String insert = statements.get(2); + assertTrue(insert.toLowerCase().contains("insert into")); + assertTrue(insert.contains("analytics_event_delete_keys")); + assertTrue(insert.contains("trackerevent")); + assertTrue(insert.contains("enrollment")); + assertTrue(insert.contains("en.programid =")); + assertTrue(insert.contains("ev.lastupdated >=")); + assertTrue(insert.contains("ev.lastupdated <")); + assertTrue( + insert.contains("(event) select ev.uid"), + () -> "Expected column-list-to-select-list pairing to match exactly, got: " + insert); + + String delete = statements.get(3); + assertTrue(delete.toLowerCase().contains("delete from")); + assertTrue(delete.toLowerCase().contains("using")); + assertTrue(delete.contains("analytics_event_delete_keys")); + assertTrue(delete.contains("ax.event=k.event")); + assertFalse(delete.contains("trackerevent")); + assertFalse(delete.contains("enrollment")); + + String drop2 = statements.get(4); + assertTrue(drop2.toLowerCase().contains("drop table")); + assertTrue(drop2.contains("analytics_event_delete_keys")); + } + + @Test + @DisplayName( + "removeUpdatedData materializes event keys natively for a single event program on Doris," + + " instead of a federated join inside DELETE") + void testRemoveUpdatedDataMaterializesKeysNativelyForSingleEvent() { + ProgramStage psA = new ProgramStage(); + psA.setId(123456); + + Program program = createProgram('A'); + program.setProgramType(ProgramType.WITHOUT_REGISTRATION); + program.setProgramStages(Set.of(psA)); + + Date lastFullTableUpdate = new DateTime(2019, 3, 1, 2, 0).toDate(); + Date lastLatestPartitionUpdate = new DateTime(2019, 3, 1, 9, 0).toDate(); + Date startTime = new DateTime(2019, 3, 1, 10, 0).toDate(); + + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder().startTime(startTime).build().withLatestPartition(); + + List> queryResp = new ArrayList<>(); + queryResp.add(Map.of("eventid", 1)); + + when(settings.getLastSuccessfulAnalyticsTablesUpdate()).thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulAnalyticsTablesUpdate(AnalyticsTableType.EVENT)) + .thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulLatestAnalyticsPartitionUpdate(AnalyticsTableType.EVENT)) + .thenReturn(lastLatestPartitionUpdate); + when(jdbcTemplate.queryForList(Mockito.anyString())).thenReturn(queryResp); + when(idObjectManager.getAllNoAcl(Program.class)).thenReturn(List.of(program)); + when(configurationService.getConfiguration()).thenReturn(configuration); + when(configuration.getDataOutputPeriodTypes()) + .thenReturn(PERIOD_TYPES.stream().collect(toUnmodifiableSet())); + + List tables = subject.getAnalyticsTables(params); + assertThat(tables, hasSize(1)); + + subject.removeUpdatedData(tables); + + ArgumentCaptor sql = ArgumentCaptor.forClass(String.class); + verify(jdbcTemplate, Mockito.times(5)).execute(sql.capture()); + List statements = sql.getAllValues(); + + String drop1 = statements.get(0); + assertTrue(drop1.toLowerCase().contains("drop table if exists")); + assertTrue(drop1.contains("analytics_event_delete_keys")); + + String create = statements.get(1); + assertTrue(create.toLowerCase().contains("create table")); + assertTrue(create.contains("analytics_event_delete_keys")); + assertTrue(create.contains("`event`")); + assertFalse(create.contains("`year`")); + + String insert = statements.get(2); + assertTrue(insert.toLowerCase().contains("insert into")); + assertTrue(insert.contains("analytics_event_delete_keys")); + assertTrue(insert.contains("singleevent")); + assertTrue(insert.contains("ev.programstageid =")); + assertTrue(insert.contains("ev.lastupdated >=")); + assertTrue(insert.contains("ev.lastupdated <")); + assertTrue( + insert.contains("(event) select ev.uid"), + () -> "Expected column-list-to-select-list pairing to match exactly, got: " + insert); + + String delete = statements.get(3); + assertTrue(delete.toLowerCase().contains("delete from")); + assertTrue(delete.toLowerCase().contains("using")); + assertTrue(delete.contains("analytics_event_delete_keys")); + assertTrue(delete.contains("ax.event=k.event")); + assertFalse(delete.contains("singleevent")); + + String drop2 = statements.get(4); + assertTrue(drop2.toLowerCase().contains("drop table")); + assertTrue(drop2.contains("analytics_event_delete_keys")); + } } diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcEventAnalyticsTableManagerTest.java b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcEventAnalyticsTableManagerTest.java index 993107de1bf3..0665c18c5b88 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcEventAnalyticsTableManagerTest.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/JdbcEventAnalyticsTableManagerTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2004-2022, University of Oslo + * Copyright (c) 2004-2026, University of Oslo * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -63,7 +63,9 @@ import static org.hisp.dhis.test.TestBase.createTrackedEntityAttribute; import static org.hisp.dhis.test.TestBase.getDate; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -1130,6 +1132,39 @@ void removeUpdatedDataTargetsLiveTable() { assertThat(sql.getValue(), not(containsString(quote(mainTableName + STAGING_TABLE_SUFFIX)))); } + @Test + @DisplayName( + "On Postgres a program without a main table gets the same continuous window as any other," + + " without checking whether its main table exists") + void latestTableOfProgramWithoutMainTableKeepsWindowOnPostgres() { + Program created = createProgram('C'); + + Date lastFullTableUpdate = new DateTime(2019, 3, 1, 2, 0).toDate(); + Date lastLatestPartitionUpdate = new DateTime(2019, 3, 1, 9, 0).toDate(); + Date startTime = new DateTime(2019, 3, 1, 10, 0).toDate(); + + AnalyticsTableUpdateParams params = + AnalyticsTableUpdateParams.newBuilder().startTime(startTime).build().withLatestPartition(); + + List> queryResp = new ArrayList<>(); + queryResp.add(Map.of("eventid", 1)); + String tableExistsSql = + sqlBuilder.tableExists(AnalyticsTable.getTableName(AnalyticsTableType.EVENT, created)); + + when(settings.getLastSuccessfulAnalyticsTablesUpdate()).thenReturn(lastFullTableUpdate); + when(settings.getLastSuccessfulLatestAnalyticsPartitionUpdate()) + .thenReturn(lastLatestPartitionUpdate); + when(jdbcTemplate.queryForList(Mockito.anyString())).thenReturn(queryResp); + when(idObjectManager.getAllNoAcl(Program.class)).thenReturn(List.of(created)); + whenConfigurationPeriodSettings(); + + List tables = subject.getAnalyticsTables(params); + assertThat(tables, hasSize(1)); + + assertNotEquals(new Date(0L), tables.get(0).getLatestTablePartition().getStartDate()); + verify(jdbcTemplate, never()).queryForList(tableExistsSql); + } + @Test @DisplayName("getRegularAnalyticsTables excludes programs listed in skipPrograms") void getRegularAnalyticsTablesExcludesSkippedPrograms() { diff --git a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/model/AnalyticsTableTest.java b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/model/AnalyticsTableTest.java index f3c7e8e91877..bae067263f88 100644 --- a/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/model/AnalyticsTableTest.java +++ b/dhis-2/dhis-services/dhis-service-analytics/src/test/java/org/hisp/dhis/analytics/table/model/AnalyticsTableTest.java @@ -215,4 +215,38 @@ void testEquals() { uniqueList.add(tableB); assertEquals(1, uniqueList.size()); } + + @Test + void testConstructorWithSortKeyAndPrimaryKey() { + List columns = + List.of( + AnalyticsTableColumn.builder().name("id").dataType(TEXT).build(), + AnalyticsTableColumn.builder().name("dx").dataType(CHARACTER_11).build()); + + AnalyticsTable table = + new AnalyticsTable( + AnalyticsTableType.DATA_VALUE, columns, List.of("dx"), List.of("id"), Logged.UNLOGGED); + + assertTrue(table.hasPrimaryKey()); + assertEquals(List.of("id"), table.getPrimaryKey()); + assertTrue(table.hasSortKey()); + assertEquals(List.of("dx"), table.getSortKey()); + } + + @Test + void testConstructorWithPrimaryKeyAndProgram() { + Program program = new Program("ProgramA", "DescriptionA"); + program.setUid("rft56ybgfek"); + + List columns = + List.of(AnalyticsTableColumn.builder().name("event").dataType(CHARACTER_11).build()); + + AnalyticsTable table = + new AnalyticsTable( + AnalyticsTableType.EVENT, columns, List.of("event"), Logged.UNLOGGED, program); + + assertTrue(table.hasPrimaryKey()); + assertEquals(List.of("event"), table.getPrimaryKey()); + assertEquals(program, table.getProgram()); + } } diff --git a/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/AbstractSqlBuilder.java b/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/AbstractSqlBuilder.java index daf157da406e..f20fd4aded4b 100644 --- a/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/AbstractSqlBuilder.java +++ b/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/AbstractSqlBuilder.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2004-2024, University of Oslo + * Copyright (c) 2004-2026, University of Oslo * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -147,6 +147,11 @@ public String tableExists(Table table) { return tableExists(table.getName()); } + @Override + public String showCreateTable(String name) { + return notSupported(); + } + @Override public String countRows(Table table) { return String.format("select count(*) as row_count from %s;", quote(table.getName())); diff --git a/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/ClickHouseSqlBuilder.java b/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/ClickHouseSqlBuilder.java index ebfa544b4066..ad8a0993c178 100644 --- a/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/ClickHouseSqlBuilder.java +++ b/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/ClickHouseSqlBuilder.java @@ -219,6 +219,16 @@ public boolean requiresIndexesForAnalytics() { return false; } + @Override + public boolean supportsContinuousAnalytics() { + return false; + } + + @Override + public boolean requiresUniqueKeyAnalyticsTables() { + return false; + } + @Override public boolean supportsPercentileCont() { return false; diff --git a/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/DorisSqlBuilder.java b/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/DorisSqlBuilder.java index 99aa77dd9b4e..8a14123b05d8 100644 --- a/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/DorisSqlBuilder.java +++ b/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/DorisSqlBuilder.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2004-2024, University of Oslo + * Copyright (c) 2004-2026, University of Oslo * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -29,6 +29,7 @@ */ package org.hisp.dhis.db.sql; +import java.util.ArrayList; import java.util.Arrays; import java.util.Comparator; import java.util.List; @@ -221,6 +222,16 @@ public boolean requiresIndexesForAnalytics() { return false; } + @Override + public boolean supportsContinuousAnalytics() { + return true; + } + + @Override + public boolean requiresUniqueKeyAnalyticsTables() { + return true; + } + @Override public boolean supportsPercentileCont() { return false; @@ -370,7 +381,7 @@ public String createTable(Table table) { // Columns if (table.hasColumns()) { - String columns = toCommaSeparated(table.getColumns(), this::toColumnString); + String columns = toCommaSeparated(getOrderedColumns(table), this::toColumnString); sql.append("(").append(columns).append(") engine = olap "); } @@ -447,6 +458,37 @@ private String generatePartitionClause(List partitions) { return partitionClause.substring(0, partitionClause.length() - 1) + ") "; } + /** + * Returns the table's columns ordered so that primary key columns appear first, in their declared + * order, followed by the remaining columns in their original order. Doris requires key columns + * (for both the unique key and duplicate key models) to be an ordered leading prefix of the table + * schema. + * + * @param table the {@link Table}. + * @return the ordered list of {@link Column}. + */ + private List getOrderedColumns(Table table) { + if (!table.hasPrimaryKey()) { + return table.getColumns(); + } + + Map columnsByName = + table.getColumns().stream().collect(Collectors.toMap(Column::getName, c -> c)); + + List keyColumns = + table.getPrimaryKey().stream().map(columnsByName::get).collect(Collectors.toList()); + + List otherColumns = + table.getColumns().stream() + .filter(c -> !table.getPrimaryKey().contains(c.getName())) + .collect(Collectors.toList()); + + List orderedColumns = new ArrayList<>(keyColumns); + orderedColumns.addAll(otherColumns); + + return orderedColumns; + } + /** * Returns a column definition string. * @@ -506,11 +548,16 @@ public String tableExists(String name) { return String.format( """ select t.table_name from information_schema.tables t \ - where t.table_schema = 'public' \ + where t.table_schema = database() \ and t.table_name = %s;""", singleQuote(name)); } + @Override + public String showCreateTable(String name) { + return String.format("show create table %s;", quote(name)); + } + /** * Doris supports indexes but relies on concurrency and compression for query performance instead * of indexes on arbitrary columns. Read more at {@link https://t.ly/AHhJ1}. diff --git a/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/PostgreSqlBuilder.java b/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/PostgreSqlBuilder.java index b60a460abb83..aa7f87c42b24 100644 --- a/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/PostgreSqlBuilder.java +++ b/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/PostgreSqlBuilder.java @@ -222,6 +222,16 @@ public boolean requiresIndexesForAnalytics() { return true; } + @Override + public boolean supportsContinuousAnalytics() { + return true; + } + + @Override + public boolean requiresUniqueKeyAnalyticsTables() { + return false; + } + @Override public boolean supportsPercentileCont() { return true; diff --git a/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/SqlBuilder.java b/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/SqlBuilder.java index 32a2ce4191ac..17758196c9fa 100644 --- a/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/SqlBuilder.java +++ b/dhis-2/dhis-support/dhis-support-sql/src/main/java/org/hisp/dhis/db/sql/SqlBuilder.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2004-2024, University of Oslo + * Copyright (c) 2004-2026, University of Oslo * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -211,6 +211,21 @@ public interface SqlBuilder { */ boolean requiresIndexesForAnalytics(); + /** + * @return true if the DBMS supports continuous (incremental latest-partition) analytics table + * updates. + */ + boolean supportsContinuousAnalytics(); + + /** + * @return true if analytics tables must be created with a real unique key for continuous + * analytics table updates to work correctly on this DBMS. When true, the natural key column + * of the affected analytics table types (e.g. {@code event}, {@code id}) is passed as the + * table's primary key so the DBMS can use it for both deduplication on insert and DELETE + * statements that join back to operational tables. + */ + boolean requiresUniqueKeyAnalyticsTables(); + /** * The percentile_cont function is a statistical function that calculates the continuous * percentile value for a given set of data. @@ -538,6 +553,16 @@ String ifThenElse( */ String tableExists(String name); + /** + * Only meaningful for engines where {@link #requiresUniqueKeyAnalyticsTables()} is true, which + * may have main tables predating the introduction of unique-key analytics tables. + * + * @param name the table name. + * @return a statement which returns the table's create statement, which can be inspected to + * determine the table's key model. + */ + String showCreateTable(String name); + /** * @param table the {@link Table}. * @return a count rows statement. diff --git a/dhis-2/dhis-support/dhis-support-sql/src/test/java/org/hisp/dhis/db/sql/DorisSqlBuilderTest.java b/dhis-2/dhis-support/dhis-support-sql/src/test/java/org/hisp/dhis/db/sql/DorisSqlBuilderTest.java index 312ae67d122e..e1f8549a6a7a 100644 --- a/dhis-2/dhis-support/dhis-support-sql/src/test/java/org/hisp/dhis/db/sql/DorisSqlBuilderTest.java +++ b/dhis-2/dhis-support/dhis-support-sql/src/test/java/org/hisp/dhis/db/sql/DorisSqlBuilderTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2004-2025, University of Oslo + * Copyright (c) 2004-2026, University of Oslo * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -32,6 +32,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.List; import org.hisp.dhis.db.model.Collation; @@ -98,6 +99,42 @@ private Table getTableD() { return new Table("immunization", columns, List.of(), sortKey, List.of(), Logged.LOGGED); } + /** + * A table whose primary key column ("id") is not the first column, mirroring the aggregate + * (JdbcAnalyticsTableManager) and completeness (JdbcCompletenessTableManager) analytics tables, + * where "id" is not a leading column. + */ + private Table getTableE() { + List columns = + List.of( + new Column("data", DataType.CHARACTER_11, Nullable.NOT_NULL), + new Column("period", DataType.VARCHAR_50, Nullable.NOT_NULL), + new Column("id", DataType.BIGINT, Nullable.NOT_NULL), + new Column("value", DataType.DOUBLE)); + + List primaryKey = List.of("id"); + + return new Table("immunization", columns, primaryKey, Logged.LOGGED); + } + + /** + * A table with a composite primary key ("id", "year") where the natural column order (`year` + * before `id`) does not match the declared key order (`id` before `year`) — mirrors the + * aggregate/completeness analytics tables, where `year` is a `FIXED_COLS` entry appended before + * the `id` column, but the Doris unique key clause must declare `id` first. + */ + private Table getTableF() { + List columns = + List.of( + new Column("year", DataType.INTEGER, Nullable.NOT_NULL), + new Column("id", DataType.BIGINT, Nullable.NOT_NULL), + new Column("data", DataType.CHARACTER_11, Nullable.NOT_NULL)); + + List primaryKey = List.of("id", "year"); + + return new Table("immunization", columns, primaryKey, Logged.LOGGED); + } + // Data types @Test @@ -105,6 +142,11 @@ void testDataTypes() { assertEquals("double", sqlBuilder.dataTypeDouble()); assertEquals("datetime(3)", sqlBuilder.dataTypeTimestamp()); assertEquals("json", sqlBuilder.dataTypeJson()); + // Doris' "string" type is unbounded and rejected as a key column on any key model (errCode = + // 2, "String Type should not be used in key column"). Callers that need a column to be a key + // (e.g. analytics table managers building a synthetic "id" key column) must use a bounded + // type (e.g. VARCHAR_255) instead of TEXT on Doris. + assertEquals("string", sqlBuilder.dataTypeText()); } @Test @@ -136,6 +178,16 @@ void testSupportsVacuum() { assertFalse(sqlBuilder.supportsVacuum()); } + @Test + void testSupportsContinuousAnalytics() { + assertTrue(sqlBuilder.supportsContinuousAnalytics()); + } + + @Test + void testRequiresUniqueKeyAnalyticsTables() { + assertTrue(sqlBuilder.requiresUniqueKeyAnalyticsTables()); + } + // Utilities @Test @@ -227,6 +279,40 @@ distributed by hash(`id`) \ assertEquals(expected, sqlBuilder.createTable(table)); } + @Test + void testCreateTableOrdersPrimaryKeyColumnFirst() { + Table table = getTableE(); + + String expected = + """ + create table `immunization` (`id` bigint not null,`data` char(11) not null,\ + `period` varchar(200) not null,`value` double null) \ + engine = olap \ + unique key (`id`) \ + distributed by hash(`id`) \ + buckets 10 \ + properties ("replication_num" = "1");"""; + + assertEquals(expected, sqlBuilder.createTable(table)); + } + + @Test + void testCreateTableOrdersCompositePrimaryKeyByDeclaredKeyOrder() { + Table table = getTableF(); + + String expected = + """ + create table `immunization` (`id` bigint not null,`year` int not null,\ + `data` char(11) not null) \ + engine = olap \ + unique key (`id`,`year`) \ + distributed by hash(`id`) \ + buckets 10 \ + properties ("replication_num" = "1");"""; + + assertEquals(expected, sqlBuilder.createTable(table)); + } + @Test void testRenameTable() { Table table = getTableA(); @@ -284,12 +370,17 @@ void testTableExists() { """ select t.table_name \ from information_schema.tables t \ - where t.table_schema = 'public' \ + where t.table_schema = database() \ and t.table_name = 'immunization';"""; assertEquals(expected, sqlBuilder.tableExists("immunization")); } + @Test + void testShowCreateTable() { + assertEquals("show create table `immunization`;", sqlBuilder.showCreateTable("immunization")); + } + @Test void testCountRows() { String expected =