Skip to content

Fix pseudo-irregular time series direct reads #1781

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
krowvin wants to merge 9 commits into
base: develop
Choose a base branch
from
+146 −52
Open

Fix pseudo-irregular time series direct reads #1781

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
97062d8
Fix pseudo-irregular time series direct reads
krowvin Jun 12, 2026
2260736
Fix pseudo-irregular direct time-series reads
krowvin Jun 20, 2026
eeca34f
Merge branch 'develop' into issue-1745-pseudo-irregular
krowvin Jun 20, 2026
5ef0b0c
Restore direct-read unit validation
krowvin Jun 20, 2026
48b5d94
Merge branch 'develop' into issue-1745-pseudo-irregular
krowvin Jun 22, 2026
744e065
Merge develop into issue-1745-pseudo-irregular
krowvin Jun 26, 2026
631dc01
Align direct TSV date predicates with retrieve_ts
krowvin Jun 26, 2026
cea2f38
Simplify direct read regularity handling
krowvin Jun 26, 2026
530030f
Fix direct read unit and version filters
krowvin Jun 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
122 changes: 83 additions & 39 deletions cwms-data-api/src/main/java/cwms/cda/data/dao/TimeSeriesDaoImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,8 @@ public class TimeSeriesDaoImpl extends JooqDao<TimeSeries> implements TimeSeries
private static final String VALUE_AT_MAX_DATE = "value_at_max_date";
private static final String CWMS_20 = "CWMS_20";
private static final String UNIT_ID = "UNIT_ID";
private static final DateTimeFormatter ORACLE_DATE_FORMATTER =
DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss");

public static final boolean OVERRIDE_PROTECTION = true;
public static final int TS_ID_MISSING_CODE = 20001;
Expand Down Expand Up @@ -187,9 +189,12 @@ public class TimeSeriesDaoImpl extends JooqDao<TimeSeries> implements TimeSeries
private final Histogram getRequestedTimeSeriesResultsReturnedHistogram;
@NotNull
private final Histogram getRequestedTimeSeriesRequestWindowMillisHistogram;
@NotNull
private final MetricRegistry metrics;

public TimeSeriesDaoImpl(DSLContext dsl, @NotNull MetricRegistry metrics) {
super(dsl);
this.metrics = metrics;

String className = this.getClass().getName();
CacheStats stats = isVersionedCache.stats();
Expand Down Expand Up @@ -240,44 +245,49 @@ private ResultQuery<Record4<Timestamp, Double, BigDecimal, Timestamp>> buildTsvD

Timestamp beginTimestamp = Timestamp.from(beginTime.toInstant());
Timestamp endTimestamp = Timestamp.from(endTime.toInstant());
String beginTimestampText = beginTimestamp.toLocalDateTime().format(ORACLE_DATE_FORMATTER);
String endTimestampText = endTimestamp.toLocalDateTime().format(ORACLE_DATE_FORMATTER);

AV_TSV_DQU view = AV_TSV_DQU.AV_TSV_DQU;

Field<BigDecimal> qualityForNormalization = DSL.nvl(
view.QUALITY_CODE.cast(BigDecimal.class),
DSL.val(BigDecimal.valueOf(5))
);

Field<BigDecimal> normalizedQuality = CWMS_TS_PACKAGE.call_NORMALIZE_QUALITY(
qualityForNormalization
).as("quality_norm");

Field<Timestamp> dateTimeField = field(name("CWMS_20", "AV_TSV_DQU", DATE_TIME), Timestamp.class);
Field<Timestamp> versionDateField = field(name("CWMS_20", "AV_TSV_DQU", VERSION_DATE), Timestamp.class);
Field<BigDecimal> qualityCode = view.QUALITY_CODE.cast(BigDecimal.class).as(QUALITY_CODE);
Field<Double> value = view.VALUE.as(VALUE);

Condition baseCondition = view.ALIASED_ITEM.isNull()
.and(view.TS_CODE.eq(tsCode))
.and(view.OFFICE_ID.eq(officeId))
.and(view.UNIT_ID.equalIgnoreCase(units))
.and(view.DATE_TIME.ge(beginTimestamp))
.and(view.DATE_TIME.le(endTimestamp))
.and(view.START_DATE.le(endTimestamp))
.and(view.END_DATE.gt(beginTimestamp));
.and(DSL.condition("{0} >= to_date({1}, 'yyyy-mm-dd\"T\"hh24:mi:ss')",
dateTimeField, DSL.val(beginTimestampText)))
.and(DSL.condition("{0} <= to_date({1}, 'yyyy-mm-dd\"T\"hh24:mi:ss')",
dateTimeField, DSL.val(endTimestampText)))
.and(view.START_DATE.isNull()
.or(DSL.condition("{0} <= to_date({1}, 'yyyy-mm-dd\"T\"hh24:mi:ss')",
view.START_DATE, DSL.val(endTimestampText))))
.and(view.END_DATE.isNull()
.or(DSL.condition("{0} > to_date({1}, 'yyyy-mm-dd\"T\"hh24:mi:ss')",
view.END_DATE, DSL.val(beginTimestampText))));

ResultQuery<Record4<Timestamp, Double, BigDecimal, Timestamp>> query;
if (versionDate != null) {
query = buildVersionedRowsQuery(
view,
dateTimeField,
versionDateField,
value,
normalizedQuality,
qualityCode,
baseCondition,
versionDate,
includeEntryDate
);
} else {
query = buildMaxVersionRowsQuery(
view,
dateTimeField,
versionDateField,
value,
normalizedQuality,
qualityCode,
baseCondition,
includeEntryDate
);
Expand Down Expand Up @@ -823,7 +833,6 @@ private TimeSeries getRequestedTimeSeriesDirect(String page, int pageSize,
if (metadata == null) {
throw new DataAccessException("Unable to resolve time series metadata for " + names);
}

long tsCode = metadata.tsCode;
String tsId = metadata.tsId;
String[] tsIdParts = splitTimeSeriesId(tsId);
Expand All @@ -842,6 +851,7 @@ private TimeSeries getRequestedTimeSeriesDirect(String page, int pageSize,
if (shouldFetchVerticalDatum(parmPart)) {
verticalDatumInfo = fetchVerticalDatumInfoSeparately(locPart, requestedUnits, office);
}
validateRequestedUnits(nativeUnits, metadataUnits);

VersionType finalDateVersionType = getDirectReadVersionType(
metadata.versionFlag, versionDate != null);
Expand All @@ -851,7 +861,7 @@ private TimeSeries getRequestedTimeSeriesDirect(String page, int pageSize,
List<TimeSeries.Record> rawRows = fetchRequestedTimeSeriesRows(tsCode, metadataOfficeId,
metadataUnits, requestParameters, includeEntryDate);
long effectiveIntervalOffset = intervalOffset;
if (isRegularSeries(intervalMinutes, intervalPart)) {
if (isRegularSeries(intervalMinutes, intervalOffset, intervalPart, isLrts)) {
effectiveIntervalOffset = resolveIntervalOffset(intervalOffset, timeZoneId, intervalPart, isLrts, rawRows);
}

Expand All @@ -868,7 +878,7 @@ private TimeSeries getRequestedTimeSeriesDirect(String page, int pageSize,
beginTime,
endTime,
metadataUnits,
resolveIntervalDuration(intervalMinutes, intervalPart),
resolveIntervalDuration(intervalMinutes, intervalOffset, intervalPart, isLrts),
verticalDatumInfo,
effectiveIntervalOffset,
timeZoneId,
Expand All @@ -886,6 +896,11 @@ private TimeSeries getRequestedTimeSeriesDirect(String page, int pageSize,

private DirectReadMetadata fetchRequestedTimeSeriesMetadataRecord(
TimeSeriesRequestParameters requestParameters) {
return fetchRequestedTimeSeriesMetadataRecord(dsl, requestParameters);
}

private DirectReadMetadata fetchRequestedTimeSeriesMetadataRecord(DSLContext metadataDsl,
TimeSeriesRequestParameters requestParameters) {
String names = requestParameters.getNames();
String office = requestParameters.getOffice();
String units = requestParameters.getUnits();
Expand Down Expand Up @@ -926,7 +941,7 @@ private DirectReadMetadata fetchRequestedTimeSeriesMetadataRecord(
var tsIdView = AV_CWMS_TS_ID.AV_CWMS_TS_ID;

SelectJoinStep<?> metadataQuery =
dsl.with(valid)
metadataDsl.with(valid)
.select(
valid.field("tscode", BigDecimal.class).as("tscode"),
valid.field("tsid", String.class).as("tsid"),
Expand Down Expand Up @@ -999,63 +1014,80 @@ private List<TimeSeries.Record> fetchRequestedTimeSeriesRows(long tsCode, String

return query.fetch(record -> {
Timestamp dateTime = record.getValue(0, Timestamp.class);
Double value = record.getValue(1, Double.class);
int qualityCode = record.getValue(2, BigDecimal.class).intValue();
Double dataValue = record.getValue(1, Double.class);
int quality = normalizeQualityCode(record.getValue(2, BigDecimal.class));
Timestamp dataEntryDate = record.getValue(3, Timestamp.class);
if (dataEntryDate != null) {
return new TimeSeries.Record(dateTime, value, qualityCode, dataEntryDate);
return new TimeSeries.Record(dateTime, dataValue, quality, dataEntryDate);
}
return new TimeSeries.Record(dateTime, value, qualityCode);
return new TimeSeries.Record(dateTime, dataValue, quality);
});
}

private static int normalizeQualityCode(BigDecimal qualityCode) {
long quality = qualityCode == null ? 5L : qualityCode.longValue();
if (quality > Integer.MAX_VALUE) {
quality -= 4_294_967_296L;
}
return (int) quality;
}

private ResultQuery<Record4<Timestamp, Double, BigDecimal, Timestamp>> buildVersionedRowsQuery(
AV_TSV_DQU view,
Field<Timestamp> dateTime,
Field<Timestamp> versionDateField,
Field<Double> value,
Field<BigDecimal> normalizedQuality,
Field<BigDecimal> qualityCode,
Condition baseCondition,
ZonedDateTime versionDate,
boolean includeEntryDate) {
Field<Timestamp> versionTimestamp = CWMS_UTIL_PACKAGE.call_TO_TIMESTAMP__2(
DSL.val(versionDate.toInstant().toEpochMilli()));
String versionTimestampText = Timestamp.from(versionDate.toInstant()).toLocalDateTime()
.format(ORACLE_DATE_FORMATTER);
Condition versionDateCondition = versionDateField.eq(versionTimestamp)
.or(DSL.condition("{0} = to_date({1}, 'yyyy-mm-dd\"T\"hh24:mi:ss')",
versionDateField, DSL.val(versionTimestampText)));
Field<Timestamp> dataEntryDateField = includeEntryDate
? view.DATA_ENTRY_DATE
: DSL.castNull(Timestamp.class).as(DATA_ENTRY_DATE);

return dsl.select(
view.DATE_TIME,
dateTime,
value,
normalizedQuality,
qualityCode,
dataEntryDateField)
.from(view)
.where(baseCondition.and(view.VERSION_DATE.eq(versionTimestamp)))
.orderBy(view.DATE_TIME.asc());
.where(baseCondition.and(versionDateCondition))
.orderBy(dateTime.asc());
}

private ResultQuery<Record4<Timestamp, Double, BigDecimal, Timestamp>> buildMaxVersionRowsQuery(
AV_TSV_DQU view,
Field<Timestamp> dateTime,
Field<Timestamp> versionDateField,
Field<Double> value,
Field<BigDecimal> normalizedQuality,
Field<BigDecimal> qualityCode,
Condition baseCondition,
boolean includeEntryDate) {
var rankedRows = dsl.select(
view.DATE_TIME.as(DATE_TIME),
dateTime.as(DATE_TIME),
value,
normalizedQuality,
qualityCode,
includeEntryDate
? view.DATA_ENTRY_DATE.as(DATA_ENTRY_DATE)
: DSL.castNull(Timestamp.class).as(DATA_ENTRY_DATE),
DSL.rowNumber()
.over(partitionBy(view.DATE_TIME)
.orderBy(view.VERSION_DATE.desc(), view.DATA_ENTRY_DATE.desc()))
.over(partitionBy(dateTime)
.orderBy(versionDateField.desc(), view.DATA_ENTRY_DATE.desc()))
.as("version_rank"))
.from(view)
.where(baseCondition)
.asTable("ranked_rows");

Field<Timestamp> dateTimeCol = rankedRows.field(DATE_TIME, Timestamp.class);
Field<Double> valueCol = rankedRows.field(VALUE, Double.class);
Field<BigDecimal> qualityCol = rankedRows.field("quality_norm", BigDecimal.class);
Field<BigDecimal> qualityCol = rankedRows.field(QUALITY_CODE, BigDecimal.class);
Field<Timestamp> dataEntryDateCol = rankedRows.field(DATA_ENTRY_DATE, Timestamp.class);
Field<Integer> versionRankCol = rankedRows.field("version_rank", Integer.class);

Expand All @@ -1065,12 +1097,18 @@ private ResultQuery<Record4<Timestamp, Double, BigDecimal, Timestamp>> buildMaxV
.orderBy(dateTimeCol.asc());
}

private void validateRequestedUnits(String nativeUnits, String requestedUnits) {
if (nativeUnits != null && requestedUnits != null) {
CWMS_UTIL_PACKAGE.call_CONVERT_UNITS(dsl.configuration(), 0.0D, nativeUnits, requestedUnits);
}
}

private List<Timestamp> fetchExpectedRegularTimes(long intervalMinutes, long intervalOffset, String timeZoneId,
String intervalPart, boolean isLrts,
TimeSeriesRequestParameters requestParameters,
List<TimeSeries.Record> rawRows) {
boolean shouldTrim = requestParameters.isShouldTrim();
if (!isRegularSeries(intervalMinutes, intervalPart)) {
if (!isRegularSeries(intervalMinutes, intervalOffset, intervalPart, isLrts)) {
return Collections.emptyList();
}
// Trimmed requests collapse to the observed data window
Expand Down Expand Up @@ -1148,11 +1186,17 @@ private long resolveIntervalOffset(long intervalOffset, String timeZoneId,
return (rawRows.get(0).getDateTime().getTime() - topOfInterval.getTime()) / TimeUnit.MINUTES.toMillis(1);
}

private boolean isRegularSeries(long intervalMinutes, String intervalPart) {
return intervalMinutes != 0L || isLocalRegularInterval(intervalPart);
private boolean isRegularSeries(long intervalMinutes, long intervalOffset, String intervalPart, boolean isLrts) {
return intervalOffset != UTC_OFFSET_IRREGULAR
&& (intervalMinutes != 0L || (isLrts && isLocalRegularInterval(intervalPart)));
}

private Duration resolveIntervalDuration(long intervalMinutes, String intervalPart) {
private Duration resolveIntervalDuration(long intervalMinutes, long intervalOffset,
String intervalPart, boolean isLrts) {
if (!isRegularSeries(intervalMinutes, intervalOffset, intervalPart, isLrts)) {
return Duration.ZERO;
}

if (intervalMinutes != 0L) {
return Duration.ofMinutes(intervalMinutes);
}
Expand Down
Loading
Loading