gCubeSystem
/
accounting-analytics-persistence-postgresql
15
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Implementing queries

This commit is contained in:
Luca Frosini 2021-03-16 15:33:34 +01:00
parent 24e31d7248
commit f8c8605523
6 changed files with 456 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

163
src/main/java/org/gcube/accounting/analytics/persistence/postgresql/AccountingPersistenceQueryPostgreSQL.java
View File

@ -3,15 +3,17 @@
*/
package org.gcube.accounting.analytics.persistence.postgresql;
import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.time.OffsetDateTime;
import java.util.Calendar;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.SortedSet;
import java.util.TreeMap;
import org.gcube.accounting.analytics.Filter;
import org.gcube.accounting.analytics.Info;
@ -27,7 +29,6 @@ import org.gcube.accounting.datamodel.aggregation.AggregatedServiceUsageRecord;
import org.gcube.accounting.datamodel.usagerecords.ServiceUsageRecord;
import org.gcube.accounting.persistence.AccountingPersistenceConfiguration;
import org.gcube.documentstore.records.AggregatedRecord;
import org.gcube.documentstore.records.Record;
import org.gcube.documentstore.records.RecordUtility;
import org.json.JSONObject;
import org.slf4j.Logger;
@ -49,13 +50,19 @@ public class AccountingPersistenceQueryPostgreSQL implements AccountingPersisten
protected Map<String,String> connectionMap;
protected static Map<Class<? extends AggregatedRecord<?, ?>>, String> recordTypes;
static {
// One Record per package is enough
RecordUtility.addRecordPackage(ServiceUsageRecord.class.getPackage());
RecordUtility.addRecordPackage(AggregatedServiceUsageRecord.class.getPackage());
recordTypes = new HashMap<>();
Map<String, Class<? extends AggregatedRecord<?,?>>> aggregatedRecords = RecordUtility.getAggregatedRecordClassesFound();
for(String typeName : aggregatedRecords.keySet()) {
try {
Class<? extends AggregatedRecord<?,?>> clz = aggregatedRecords.get(typeName);
UsageRecordToDBMapping.getRecordToDB(clz);
} catch (Exception e) {
new RuntimeException(e);
}
}
}
@Override
@ -66,26 +73,7 @@ public class AccountingPersistenceQueryPostgreSQL implements AccountingPersisten
this.connectionMap = new HashMap<>();
}
protected static String getRecordTypeByClass(Class<? extends AggregatedRecord<?, ?>> clz) {
try {
Record r = clz.newInstance();
return r.getRecordType();
}catch (Exception e) {
String type = clz.getSimpleName();
type = type.replace("Abstract", "");
type = type.replace("Aggregated", "");
return type;
}
}
protected static String getRecordType(Class<? extends AggregatedRecord<?, ?>> clz) {
String type = recordTypes.get(clz);
if(type == null) {
type = getRecordTypeByClass(clz);
recordTypes.put(clz, type);
}
return type;
}
@Override
public SortedMap<Calendar, Info> getTimeSeries(Class<? extends AggregatedRecord<?, ?>> clz,
@ -95,111 +83,42 @@ public class AccountingPersistenceQueryPostgreSQL implements AccountingPersisten
return null;
}
protected void appendString(StringBuffer query, String string) {
query.append("'");
query.append(string);
query.append("'");
}
protected void appendValue(StringBuffer values, Serializable serializable) {
if(serializable instanceof Number) {
values.append(serializable.toString());
return;
}
if(serializable instanceof Calendar) {
Calendar calendar = (Calendar) serializable;
SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DATETIME_PATTERN);
String date = simpleDateFormat.format(calendar.getTime());
appendString(values, date);
return;
}
if(serializable instanceof Enum) {
Enum<?> e = (Enum<?>) serializable;
appendString(values, e.name());
return;
}
// String, URI etc
appendString(values, serializable.toString());
}
protected void appendKey(StringBuffer query, String fieldName) {
int lenght = fieldName.length();
boolean lastLowerCase = true;
for (int i=0; i<lenght; i++) {
Character ch = fieldName.charAt(i); /*traversing String one by one*/
if (Character.isUpperCase(ch)) {
if(lastLowerCase) {
query.append("_");
}
lastLowerCase = false;
}else {
lastLowerCase = true;
}
query.append(Character.toLowerCase(ch));
}
}
/* this solution does not works because postgres deos not support upper case charaters
protected void appendKeyAs(StringBuffer query, String fieldName) {
appendKey(query, fieldName);
query.append(" AS ");
query.append(fieldName);
}
*/
protected String getNoContextTimeSeriesQuery(Class<? extends AggregatedRecord<?, ?>> clz,
TemporalConstraint temporalConstraint, List<Filter> filters)
throws DuplicatedKeyFilterException, KeyException, ValueException, Exception {
Set<String> aggregatedField = clz.newInstance().getAggregatedFields();
StringBuffer query = new StringBuffer();
query.append("SELECT ");
/*
* We want StartTime as first field for this reason has been
* excluded from cycle
*/
appendKey(query, AggregatedRecord.START_TIME);
for(String fieldName : aggregatedField) {
switch (fieldName) {
case AggregatedRecord.START_TIME:
case AggregatedRecord.END_TIME:
case AggregatedRecord.AGGREGATED:
continue;
default:
query.append(", ");
appendKey(query, fieldName);
break;
}
}
query.append(" FROM ");
query.append(getRecordType(clz));
query.append(" WHERE ");
appendKey(query, AggregatedRecord.START_TIME);
query.append(" > ");
appendValue(query, temporalConstraint.getAlignedStartTime());
query.append(" AND ");
appendKey(query, AggregatedRecord.START_TIME);
query.append(" < ");
appendValue(query, temporalConstraint.getAlignedEndTime());
return query.toString();
protected Calendar getCalendar(OffsetDateTime offsetDateTime) {
Calendar calendar = Calendar.getInstance();
long epochMillis = offsetDateTime.toInstant().toEpochMilli();
calendar.setTimeInMillis(epochMillis);
return calendar;
}
@Override
public SortedMap<Calendar, Info> getNoContextTimeSeries(Class<? extends AggregatedRecord<?, ?>> clz,
TemporalConstraint temporalConstraint, List<Filter> filters)
throws DuplicatedKeyFilterException, KeyException, ValueException, Exception {
Connection connection = null;
Statement statement = connection.createStatement();
SortedMap<Calendar, Info> result = new TreeMap<>();
Query query = new Query(clz);
query.setTemporalConstraint(temporalConstraint);
query.setFilters(filters);
String sql = query.getNoContextTimeSeriesQuery();
List<String> requestedTableField = query.getRequestedTableField();
ResultSet resultSet = statement.executeQuery(sql);
while (resultSet.next()) {
OffsetDateTime offsetDateTime = resultSet.getObject(Query.DATE_OF_TIMESERIES_AS_FIELD, OffsetDateTime.class);
Calendar calendar = getCalendar(offsetDateTime);
}

227
src/main/java/org/gcube/accounting/analytics/persistence/postgresql/Query.java Normal file
View File

@ -0,0 +1,227 @@
package org.gcube.accounting.analytics.persistence.postgresql;
import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import java.util.Set;
import org.gcube.accounting.analytics.Filter;
import org.gcube.accounting.analytics.TemporalConstraint;
import org.gcube.accounting.analytics.TemporalConstraint.AggregationMode;
import org.gcube.accounting.analytics.TemporalConstraint.CalendarEnum;
import org.gcube.accounting.datamodel.aggregation.AggregatedServiceUsageRecord;
import org.gcube.documentstore.records.AggregatedRecord;
public class Query {
public static final String DATETIME_PATTERN = "yyyy-MM-dd HH:mm:ss.SSS Z";
public static final String DATE_OF_TIMESERIES_AS_FIELD = "startTime";
private Class<? extends AggregatedRecord<?, ?>> clz;
private final RecordToDBMapper recordToDBMapper;
private List<String> requestedTableField;
private StringBuffer stringBuffer;
protected TemporalConstraint temporalConstraint;
protected List<Filter> filters;
protected Set<String> contexts;
public Query(Class<? extends AggregatedRecord<?, ?>> clz) throws Exception {
this.clz = clz;
this.recordToDBMapper = UsageRecordToDBMapping.getRecordToDB(clz);
}
public List<String> getRequestedTableField() {
return requestedTableField;
}
public void setTemporalConstraint(TemporalConstraint temporalConstraint) {
this.temporalConstraint = temporalConstraint;
}
public void setFilters(List<Filter> filters) {
this.filters = filters;
}
public void setContexts(Set<String> contexts) {
this.contexts = contexts;
}
public RecordToDBMapper getRecordToDBMapper() {
return recordToDBMapper;
}
private void appendString(String string) {
stringBuffer.append("'");
stringBuffer.append(string);
stringBuffer.append("'");
}
protected String getTableField(String fieldName) {
return recordToDBMapper.getTableField(fieldName);
}
protected void appendTableField(String fieldName) {
stringBuffer.append(getTableField(fieldName));
}
protected void appendValue(Serializable serializable) {
if(serializable instanceof Number) {
stringBuffer.append(serializable.toString());
return;
}
if(serializable instanceof Calendar) {
Calendar calendar = (Calendar) serializable;
SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DATETIME_PATTERN);
String date = simpleDateFormat.format(calendar.getTime());
appendString(date);
return;
}
if(serializable instanceof Enum) {
Enum<?> e = (Enum<?>) serializable;
appendString(e.name());
return;
}
// String, URI etc
appendString(serializable.toString());
}
protected void addFilters() {
if(filters!=null && filters.size()>0) {
// The first filter if the time_bucket
stringBuffer.append(" AND ");
boolean first = true;
for(Filter filter : filters) {
if(first) {
first = false;
}else {
stringBuffer.append(" AND ");
}
appendTableField(filter.getKey());
stringBuffer.append("=");
appendValue(filter.getValue());
}
}
}
protected void addEmittedFields() throws Exception {
Set<String> aggregatedField = clz.newInstance().getAggregatedFields();
for(String fieldName : aggregatedField) {
String dbField = getTableField(fieldName);
switch (fieldName) {
case AggregatedRecord.START_TIME:
case AggregatedRecord.END_TIME:
case AggregatedRecord.AGGREGATED:
continue;
case AggregatedRecord.OPERATION_COUNT:
stringBuffer.append(", SUM(");
stringBuffer.append(dbField);
stringBuffer.append(") AS ");
break;
// WEIGHTED AVERAGE
case AggregatedServiceUsageRecord.DURATION:
stringBuffer.append(", SUM(");
stringBuffer.append(dbField);
stringBuffer.append("*");
appendTableField(AggregatedRecord.OPERATION_COUNT);
stringBuffer.append(")/SUM(");
appendTableField(AggregatedRecord.OPERATION_COUNT);
stringBuffer.append(") AS ");
break;
case AggregatedServiceUsageRecord.MAX_INVOCATION_TIME:
stringBuffer.append(", MAX(");
stringBuffer.append(dbField);
stringBuffer.append(") AS ");
break;
case AggregatedServiceUsageRecord.MIN_INVOCATION_TIME:
stringBuffer.append(", MIN(");
stringBuffer.append(dbField);
stringBuffer.append(") AS ");
break;
default:
stringBuffer.append(", ");
break;
}
stringBuffer.append(dbField);
requestedTableField.add(dbField);
}
}
protected String getTimeBucketCalendarInterval(AggregationMode aggregationMode) {
CalendarEnum calendarEnum = CalendarEnum.values()[aggregationMode.ordinal()];
return calendarEnum.name().toLowerCase();
}
protected void addTimeBucket() {
stringBuffer.append("time_bucket('1 ");
String calendarInterval = getTimeBucketCalendarInterval(temporalConstraint.getAggregationMode());
stringBuffer.append(calendarInterval);
stringBuffer.append("',");
appendTableField(AggregatedRecord.START_TIME);
stringBuffer.append(") AS ");
stringBuffer.append(DATE_OF_TIMESERIES_AS_FIELD);
requestedTableField.add(DATE_OF_TIMESERIES_AS_FIELD);
}
private void newQuery() {
stringBuffer = new StringBuffer();
requestedTableField = new ArrayList<>();
stringBuffer.append("SELECT ");
}
protected void addTemporalConstraintToQuery() {
stringBuffer.append(" WHERE ");
String tableField = getTableField(AggregatedRecord.START_TIME);
stringBuffer.append(tableField);
stringBuffer.append(" > ");
appendValue(temporalConstraint.getAlignedStartTime());
stringBuffer.append(" AND ");
stringBuffer.append(tableField);
stringBuffer.append(" < ");
appendValue(temporalConstraint.getAlignedEndTime());
}
protected void addGropuBy() {
stringBuffer.append(" GROUP BY ");
stringBuffer.append(DATE_OF_TIMESERIES_AS_FIELD);
}
protected void addOrderBy() {
stringBuffer.append(" ORDER BY ");
stringBuffer.append(DATE_OF_TIMESERIES_AS_FIELD);
stringBuffer.append(" ASC");
}
public String getNoContextTimeSeriesQuery() throws Exception {
newQuery();
addTimeBucket();
addEmittedFields();
stringBuffer.append(" FROM ");
stringBuffer.append(recordToDBMapper.getTableName());
addTemporalConstraintToQuery();
addFilters();
addGropuBy();
addOrderBy();
return stringBuffer.toString();
}
}

94
src/main/java/org/gcube/accounting/analytics/persistence/postgresql/RecordToDBMapper.java Normal file
View File

@ -0,0 +1,94 @@
package org.gcube.accounting.analytics.persistence.postgresql;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.gcube.documentstore.records.AggregatedRecord;
import org.gcube.documentstore.records.Record;
public class RecordToDBMapper {
protected static String getRecordTypeByClass(Class<? extends AggregatedRecord<?, ?>> clz) {
try {
Record r = clz.newInstance();
return r.getRecordType();
}catch (Exception e) {
String type = clz.getSimpleName();
type = type.replace("Abstract", "");
type = type.replace("Aggregated", "");
return type;
}
}
protected static String getKey(String fieldName) {
StringBuffer stringBuffer = new StringBuffer();
int lenght = fieldName.length();
boolean lastLowerCase = true;
for (int i=0; i<lenght; i++) {
Character ch = fieldName.charAt(i); /*traversing String one by one*/
if (Character.isUpperCase(ch)) {
if(lastLowerCase) {
stringBuffer.append("_");
}
lastLowerCase = false;
}else {
lastLowerCase = true;
}
stringBuffer.append(Character.toLowerCase(ch));
}
return stringBuffer.toString();
}
protected final Class<? extends AggregatedRecord<?, ?>> clz;
protected final String typeName;
protected final String tableName;
protected final Map<String, String> tableFieldToUsageRecordField;
protected final Map<String, String> usageRecordFieldToTableField;
public RecordToDBMapper(Class<? extends AggregatedRecord<?, ?>> clz) throws Exception {
this.clz = clz;
this.typeName = getRecordTypeByClass(clz);
this.tableName = typeName.toLowerCase();
this.tableFieldToUsageRecordField = new HashMap<>();
this.usageRecordFieldToTableField = new HashMap<>();
mapFields();
}
public RecordToDBMapper(String typeName, Class<? extends AggregatedRecord<?, ?>> clz) throws Exception {
this.clz = clz;
this.typeName = typeName;
this.tableName = typeName.toLowerCase();
this.tableFieldToUsageRecordField = new HashMap<>();
this.usageRecordFieldToTableField = new HashMap<>();
mapFields();
}
protected void mapFields() throws Exception {
Set<String> requiredFields = clz.newInstance().getRequiredFields();
for(String usageRecordField : requiredFields) {
String dbField = getKey(usageRecordField);
tableFieldToUsageRecordField.put(dbField, usageRecordField);
usageRecordFieldToTableField.put(usageRecordField, dbField);
}
}
public String getTableField(String usageRecordField) {
return usageRecordFieldToTableField.get(usageRecordField);
}
public String getUsageRecordField(String tableField) {
return tableFieldToUsageRecordField.get(tableField);
}
public String getTypeName() {
return typeName;
}
public String getTableName() {
return tableName;
}
}

25
src/main/java/org/gcube/accounting/analytics/persistence/postgresql/UsageRecordToDBMapping.java Normal file
View File

@ -0,0 +1,25 @@
package org.gcube.accounting.analytics.persistence.postgresql;
import java.util.HashMap;
import java.util.Map;
import org.gcube.documentstore.records.AggregatedRecord;
public class UsageRecordToDBMapping {
protected final static Map<Class<? extends AggregatedRecord<?, ?>>, RecordToDBMapper> classToRecordToDBMapper;
static {
classToRecordToDBMapper = new HashMap<>();
}
public static synchronized RecordToDBMapper getRecordToDB(Class<? extends AggregatedRecord<?, ?>> clz) throws Exception {
RecordToDBMapper recordToDBMapper = classToRecordToDBMapper.get(clz);
if(recordToDBMapper==null) {
recordToDBMapper = new RecordToDBMapper(clz);
classToRecordToDBMapper.put(clz, recordToDBMapper);
}
return recordToDBMapper;
}
}

14
src/test/java/org/gcube/accounting/analytics/persistence/postgresql/AccountingPersistenceQueryPostgreSQLTest.java
View File

@ -3,13 +3,17 @@
*/
package org.gcube.accounting.analytics.persistence.postgresql;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import org.gcube.accounting.analytics.Filter;
import org.gcube.accounting.analytics.TemporalConstraint;
import org.gcube.accounting.analytics.TemporalConstraint.AggregationMode;
import org.gcube.accounting.analytics.exception.DuplicatedKeyFilterException;
import org.gcube.accounting.analytics.exception.KeyException;
import org.gcube.accounting.analytics.exception.ValueException;
import org.gcube.accounting.datamodel.UsageRecord;
import org.gcube.accounting.datamodel.aggregation.AggregatedServiceUsageRecord;
import org.junit.Test;
import org.slf4j.Logger;
@ -37,9 +41,15 @@ public class AccountingPersistenceQueryPostgreSQLTest extends ContextTest {
entTimeCalendar.set(Calendar.HOUR_OF_DAY, 16);
entTimeCalendar.set(Calendar.MINUTE, 17);
List<Filter> filters = new ArrayList<>();
Filter filter = new Filter(UsageRecord.CONSUMER_ID, "name.surname");
filters.add(filter);
TemporalConstraint temporalConstraint = new TemporalConstraint(startTimeCalendar.getTimeInMillis(), entTimeCalendar.getTimeInMillis(), AggregationMode.MINUTELY);
AccountingPersistenceQueryPostgreSQL accountingPersistenceQueryPostgreSQL = new AccountingPersistenceQueryPostgreSQL();
String ret = accountingPersistenceQueryPostgreSQL.getNoContextTimeSeriesQuery(AggregatedServiceUsageRecord.class, temporalConstraint, null);
Query query = new Query(AggregatedServiceUsageRecord.class);
query.setTemporalConstraint(temporalConstraint);
query.setFilters(filters);
String ret = query.getNoContextTimeSeriesQuery();
logger.debug(ret);
}

57
src/test/java/org/gcube/accounting/analytics/persistence/postgresql/QueryTest.java Normal file
View File

@ -0,0 +1,57 @@
/**
*
*/
package org.gcube.accounting.analytics.persistence.postgresql;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import org.gcube.accounting.analytics.Filter;
import org.gcube.accounting.analytics.TemporalConstraint;
import org.gcube.accounting.analytics.TemporalConstraint.AggregationMode;
import org.gcube.accounting.analytics.exception.DuplicatedKeyFilterException;
import org.gcube.accounting.analytics.exception.KeyException;
import org.gcube.accounting.analytics.exception.ValueException;
import org.gcube.accounting.datamodel.UsageRecord;
import org.gcube.accounting.datamodel.aggregation.AggregatedServiceUsageRecord;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author Luca Frosini (ISTI - CNR)
*/
public class QueryTest extends ContextTest {
private static final Logger logger = LoggerFactory.getLogger(QueryTest.class);
@Test
public void testGetNoContextTimeSeriesQuery() throws DuplicatedKeyFilterException, KeyException, ValueException, Exception {
logger.debug("test");
Calendar startTimeCalendar = Calendar.getInstance();
startTimeCalendar.set(Calendar.MONTH, Calendar.MARCH);
startTimeCalendar.set(Calendar.DAY_OF_MONTH, 15);
startTimeCalendar.set(Calendar.HOUR_OF_DAY, 16);
startTimeCalendar.set(Calendar.MINUTE, 15);
Calendar entTimeCalendar = Calendar.getInstance();
entTimeCalendar.set(Calendar.MONTH, Calendar.MARCH);
entTimeCalendar.set(Calendar.DAY_OF_MONTH, 15);
entTimeCalendar.set(Calendar.HOUR_OF_DAY, 16);
entTimeCalendar.set(Calendar.MINUTE, 17);
List<Filter> filters = new ArrayList<>();
Filter filter = new Filter(UsageRecord.CONSUMER_ID, "name.surname");
filters.add(filter);
TemporalConstraint temporalConstraint = new TemporalConstraint(startTimeCalendar.getTimeInMillis(), entTimeCalendar.getTimeInMillis(), AggregationMode.MINUTELY);
Query query = new Query(AggregatedServiceUsageRecord.class);
query.setTemporalConstraint(temporalConstraint);
query.setFilters(filters);
String ret = query.getNoContextTimeSeriesQuery();
logger.debug(ret);
}
}