return DB_CURSOR_ERROR;
}
- buf = cursor->tuple;
-
if (IS_INVALID_STORAGE_ROW(cursor)) {
DB_LOG_E("invalid storage row id\n");
return DB_CURSOR_ERROR;
}
- offset = cursor->current_storage_row * cursor->storage_row_length + cursor->attr_map[col].from_offset;
-
- fd = storage_open(cursor->name, O_RDONLY);
- if (fd < 0) {
- DB_LOG_E("failed to open storage %s\n", cursor->name);
- return DB_CURSOR_ERROR;
- }
- storage_read_from(fd, buf, offset, cursor->attr_map[col].from_data_size);
- storage_close(fd);
+ buf = cursor->tuple;
memcpy(attr.name, cursor->attr_map[col].name, sizeof(attr.name));
attr.domain = cursor->attr_map[col].domain;
- attr.element_size = cursor->attr_map[col].from_data_size;
+ attr.element_size = cursor->attr_map[col].data_size;
+
+ if (cursor->attr_map[col].valuetype == AGGREGATE_VALUE) {
+ /* If the type of value is aggregate value, we don't need to read storage.
+ Because aggregate result is already calculated and stored in buffer. */
+ buf += cursor->attr_map[col].offset;
+ } else {
+ /* Otherwise, Read tuple value from storage. */
+ offset = cursor->current_storage_row * cursor->storage_row_length + cursor->attr_map[col].offset;
+ fd = storage_open(cursor->name, O_RDONLY);
+ if (fd < 0) {
+ DB_LOG_E("failed to open storage %s\n", cursor->name);
+ return DB_CURSOR_ERROR;
+ }
+ storage_read_from(fd, buf, offset, attr.element_size);
+ storage_close(fd);
+ }
return db_phy_to_value(value, &attr, buf);
}
for (i = 0; i < attribute_count; i++) {
memset(cursor->attr_map[i].name, 0, sizeof(cursor->attr_map[i].name));
memcpy(cursor->attr_map[i].name, attr_map_ptr->to_attr->name, sizeof(attr_map_ptr->to_attr->name));
- cursor->attr_map[i].from_data_size = attr_map_ptr->from_attr->element_size;
- cursor->attr_map[i].from_offset = attr_map_ptr->from_offset;
cursor->attr_map[i].domain = attr_map_ptr->to_attr->domain;
+ cursor->attr_map[i].valuetype = attr_map_ptr->valuetype;
+ if (cursor->attr_map[i].valuetype == AGGREGATE_VALUE) {
+ cursor->attr_map[i].data_size = attr_map_ptr->to_attr->element_size;
+ cursor->attr_map[i].offset = attr_map_ptr->to_offset;
+ } else {
+ cursor->attr_map[i].data_size = attr_map_ptr->from_attr->element_size;
+ cursor->attr_map[i].offset = attr_map_ptr->from_offset;
+ }
attr_map_ptr++;
}
static db_result_t aggregate(attribute_t *attr, attribute_value_t *value, tuple_id_t count)
{
long long_value;
- static long sum_value;
+ double sum_value;
switch (value->domain) {
case DOMAIN_INT:
return DB_TYPE_ERROR;
}
+ sum_value = 0;
switch (attr->aggregator) {
case AQL_COUNT:
attr->aggregation_value++;
break;
case AQL_MEAN:
if (count == 1) {
- sum_value = 0;
attr->aggregation_value = (double)long_value;
} else {
- sum_value += long_value;
- attr->aggregation_value = (double)sum_value / count;
+ sum_value = (double) (attr->aggregation_value * (count - 1));
+ attr->aggregation_value = (double) ((sum_value + long_value) / count);
}
break;
case AQL_MAX:
break;
}
+ DB_LOG_D("DB: aggregation value of attribute %s is %f.\n", attr->name, attr->aggregation_value);
+
return DB_OK;
}
attr_map_ptr->from_offset = offset;
attr_map_ptr->to_offset = size_sum;
size_sum += to_attr->element_size;
+
+ if (to_attr->aggregator != 0) {
+ attr_map_ptr->valuetype = AGGREGATE_VALUE;
+ } else {
+ attr_map_ptr->valuetype = NORMAL_VALUE;
+ }
+
attr_map_ptr++;
to_attr = to_attr->next;
}
/* Check whether the given predicate is true for this tuple. */
if ((*handle)->lvm_instance == NULL || lvm_execute((*handle)->lvm_instance) == TRUE) {
(*handle)->current_row++;
+
if ((*handle)->adt_flags & AQL_FLAG_AGGREGATE) {
for (attr_map_ptr = (*handle)->attr_map; attr_map_ptr < attr_map_end; attr_map_ptr++) {
from_ptr = row + attr_map_ptr->from_offset;
if (DB_ERROR(result)) {
goto errout;
}
- if (row != NULL) {
- free(row);
- }
- return DB_GOT_ROW;
}
}
result_attr = attr_map_ptr->to_attr;
to_ptr = result_row + attr_map_ptr->to_offset;
- snprintf(aggr_buf, sizeof(aggr_buf), "%.8f", result_attr->aggregation_value);
+ snprintf(aggr_buf, sizeof(aggr_buf), "%f", result_attr->aggregation_value);
from_ptr = (unsigned char *)aggr_buf;
memcpy(to_ptr, from_ptr, sizeof(aggr_buf));
}
- (*handle)->current_row = 1;
- (*handle)->adt_flags &= ~AQL_FLAG_AGGREGATE; /* Stop the aggregation. */
+ /* Copy aggregated result to tuple in cursor */
+ memcpy(cursor->tuple, result_row, sizeof(cursor->tuple));
+
+ (*handle)->current_row = 0;
+ (*handle)->adt_flags &= ~AQL_FLAG_AGGREGATE; /* Stop the aggregation. */
+
+ result = cursor_data_add(cursor, (*handle)->current_row);
+ if (DB_ERROR(result)) {
+ goto errout;
+ }
+ cursor->total_rows = 1;
+
if (row != NULL) {
free(row);
}
- return DB_GOT_ROW;
+ return DB_FINISHED;
errout:
if (row != NULL) {
};
typedef enum db_direction_e db_direction_t;
+enum db_value_type_e {
+ NORMAL_VALUE = 0,
+ AGGREGATE_VALUE = 1
+};
+typedef enum db_value_type_e db_value_type_t;
/*
* A relation consists of a name, a set of domains, a set of indexes,
struct cursor_data_map_s {
char name[ATTRIBUTE_NAME_LENGTH + 1];
domain_t domain;
- unsigned from_data_size;
- unsigned from_offset;
+ db_value_type_t valuetype;
+ unsigned data_size;
+ unsigned offset;
};
typedef struct cursor_data_map_s cursor_data_map_t;
attribute_t *to_attr;
unsigned from_offset;
unsigned to_offset;
+ db_value_type_t valuetype;
};
typedef struct source_dest_map_s source_dest_map_t;