| // @@@ START COPYRIGHT @@@ |
| // |
| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| // |
| // @@@ END COPYRIGHT @@@ |
| |
| #include "SqlUdrPredefTimeSeries.h" |
| #include <limits> |
| |
| // sample invocation of the TIMESERIES TMUDF: |
| // |
| // SELECT time_slice, a, b, c_LLI -- note the suffix applied by the UDF |
| // FROM UDF(timeseries( |
| // -- input table with partition by/order by |
| // TABLE(SELECT a,b,c FROM t PARTITION BY a ORDER BY b), |
| // 'TIME_SLICE', -- name of time slice column |
| // INTERVAL '30' SECOND, -- length of time slice |
| // 'C', -- name of first column to interpolate |
| // 'LLI')) -- instructions how to interpolate |
| // -- (Last value, Linear, ignore nulls) |
| // -- optionally followed by more pairs of columns and instructions |
| // |
| // Allowed instructions (lower or upper case): |
| // FC[I] First value, constant interpolation |
| // LC[I] Last value, constant interpolation |
| // FL[I] First value, linear interpolation |
| // LL[I] Last value, linear interpolation |
| |
| // factory method |
| |
| extern "C" UDR * TRAF_CPP_TIMESERIES() |
| { |
| return new TimeSeries(); |
| } |
| |
| void TimeSeries::describeParamsAndColumns(UDRInvocationInfo &info) |
| { |
| InternalColumns internalCols(info); |
| |
| // create PARTITION BY output columns, one passthru column |
| // for every column that appears in PARTITION BY |
| const PartitionInfo &part = info.in().getQueryPartitioning(); |
| int numPartCols = part.getNumEntries(); |
| |
| for (int pc=0; pc<numPartCols; pc++) |
| info.addPassThruColumns(0, part.getColumnNum(pc), part.getColumnNum(pc)); |
| |
| // since we work locally in a partition, set the function type |
| // of this TMUDF to REDUCER |
| info.setFuncType(UDRInvocationInfo::REDUCER); |
| |
| // produce the time column, it has the same type as the |
| // ORDER BY column that defines the input time value |
| // and its name is specified by parameter 0 |
| const TypeInfo &timeType = |
| info.in().getColumn(internalCols.getTimeSliceInColNum()).getType(); |
| |
| info.out().addColumn(ColumnInfo(info.par().getString(0).c_str(), |
| timeType)); |
| |
| // produce aggregate columns |
| for (int a=0; a<internalCols.getNumAggrCols(); a++) |
| { |
| TimeSeriesAggregate tsa = internalCols.getAggrColumn(a); |
| std::string outColName(info.par().getString(2*a + 2)); |
| TypeInfo inColType( |
| info.in().getColumn(tsa.getInputColNum()).getType()); |
| |
| // append suffix to input column name to form the output column |
| // name, make those all capitals to avoid delimited identifiers |
| outColName += "_"; |
| if (tsa.isFirstVal()) |
| outColName += "F"; |
| else |
| outColName += "L"; |
| if (tsa.isConstInterpol()) |
| outColName += "C"; |
| else |
| outColName += "L"; |
| if (tsa.isIgnoreNulls()) |
| outColName += "I"; |
| |
| if (tsa.isConstInterpol()) |
| { |
| // add a column with the same data type as the original |
| // column, but make it nullable if it isn't already |
| inColType.setNullable(true); |
| info.out().addColumn(ColumnInfo(outColName.c_str(), inColType)); |
| } |
| else |
| // add a "DOUBLE" output column to allow interpolation |
| info.out().addColumn(ColumnInfo( |
| outColName.c_str(), |
| TypeInfo(TypeInfo::DOUBLE_PRECISION, |
| 0, |
| true))); |
| } |
| |
| // add formal parameters with types that match the actual ones |
| for (int p=0; p<info.par().getNumColumns(); p++) |
| { |
| char parName[20]; |
| |
| snprintf(parName, sizeof(parName), "PAR_%d", p); |
| info.addFormalParameter(ColumnInfo(parName, |
| info.par().getColumn(p).getType())); |
| } |
| } |
| |
| void TimeSeries::processData(UDRInvocationInfo &info, |
| UDRPlanInfo &plan) |
| { |
| InternalColumns internalCols(info); |
| bool haveMoreRows = false; |
| bool haveRowFromPreviousLoop = true; // will read one below |
| |
| // this will exit right away if there are no rows, |
| // doing this make the logic below a bit simpler |
| haveMoreRows = getNextRow(info); |
| |
| while (haveMoreRows) |
| { |
| bool inSamePartition = true; |
| |
| // read all rows of one partition, |
| // plus the next row or end of data |
| while (inSamePartition) |
| { |
| if (!haveRowFromPreviousLoop) |
| { |
| // try to read a new row |
| if (getNextRow(info)) |
| { |
| // is this row still from the same partition? |
| if ((inSamePartition = internalCols.isSamePartition()) == false) |
| haveRowFromPreviousLoop = true; |
| } |
| else |
| { |
| // we reached end-of-data, exit this loop |
| // and emit the last partition |
| haveMoreRows = false; |
| inSamePartition = false; |
| } |
| } |
| else |
| { |
| // consume the row we read in the last iteration |
| // or in the initialization code, this indicates |
| // the start of a new partition |
| |
| // prepare for the next partition |
| internalCols.initializePartition(); |
| haveRowFromPreviousLoop = false; |
| } |
| |
| if (inSamePartition) |
| { |
| // remember all the columns of this row for |
| // the computed time slice number |
| internalCols.readInputCols(); |
| } |
| } // read rows of one partition... |
| |
| // we read all rows from one partition (plus one more, maybe), |
| // now emit the rows |
| |
| // prepare to emit all time slices for this partition |
| internalCols.finalizePartition(); |
| |
| int numTimeSlices = internalCols.getNumTimeSlices(); |
| |
| // loop over time slices and emit rows |
| for (int s=0; s<numTimeSlices; s++) |
| { |
| // generate an output record for one time slice of |
| // one partition |
| internalCols.setOutputCols(s); |
| |
| // emit a row for one time slice |
| emitRow(info); |
| } |
| } // emitPartition |
| } |
| |
| template<class T> |
| NullableTimedValue<double> NullableTimedValue<T>::interpolateLinear( |
| time_t t, const NullableTimedValue<T> &nextHigherVal) |
| { |
| // return a NULL value if t is outside the range of |
| // "this" and "nextHigherVal" or if "this" or |
| // "nextHigherVal" is a NULL value |
| if (t < t_ || |
| t > nextHigherVal.t_ || |
| isNull_ || |
| nextHigherVal.isNull_) |
| return NullableTimedValue<double>(true, t, 0.0); |
| |
| // this should not be common, but avoid division by zero for this unlikely case |
| if (t_ == nextHigherVal.t_) |
| return NullableTimedValue<double>(false, t, (v_ + nextHigherVal.v_)/2); |
| |
| return NullableTimedValue<double>( |
| false, |
| t, |
| static_cast<double>(v_) + |
| (static_cast<double>(t)-t_)/(nextHigherVal.t_-t_) * (nextHigherVal.v_ - v_)); |
| } |
| |
| TimeSeriesAggregate::TimeSeriesAggregate(const TupleInfo &inTup, |
| const TupleInfo &outTup, |
| int inputColNum, |
| int outputColNum, |
| bool isFirstVal, |
| bool isConstInterpol, |
| bool isIgnoreNulls) : |
| inTup_(inTup), |
| outTup_(outTup), |
| inputColNum_(inputColNum), |
| outputColNum_(outputColNum), |
| isFirstVal_(isFirstVal), |
| isConstInterpol_(isConstInterpol), |
| isIgnoreNulls_(isIgnoreNulls), |
| currIx_(0) |
| { |
| // use a long data type for constant interpolation when |
| // the underlying column is an exact numeric, otherwise |
| // use a double |
| useLong_ = (isConstInterpol && |
| inTup_.getColumn(inputColNum_).getType().getSQLTypeSubClass() |
| == TypeInfo::EXACT_NUMERIC_TYPE); |
| } |
| |
| TimeSeriesAggregate::~TimeSeriesAggregate() |
| { |
| } |
| |
| void TimeSeriesAggregate::initPartition() |
| { |
| if (useLong_) |
| lValues_.clear(); |
| else |
| dValues_.clear(); |
| |
| currIx_ = currSliceNum_ = -1; |
| entriesForThisTimeSlice_ = 0; |
| } |
| |
| int TimeSeriesAggregate::numEntries() |
| { |
| if (useLong_) |
| return lValues_.size(); |
| else |
| return dValues_.size(); |
| } |
| |
| time_t TimeSeriesAggregate::getTime(int entry) |
| { |
| if (useLong_) |
| return lValues_[entry].getTime(); |
| else |
| return dValues_[entry].getTime(); |
| } |
| |
| void TimeSeriesAggregate::readInputCol(time_t t, int sliceNum) |
| { |
| long lVal = 0; |
| double dVal = 0.0; |
| bool wasNull; |
| |
| if (useLong_) |
| lVal = inTup_.getLong(inputColNum_); |
| else |
| dVal = inTup_.getDouble(inputColNum_); |
| wasNull = inTup_.wasNull(); |
| |
| // check whether we should ignore a NULL value |
| if (!(wasNull && isIgnoreNulls_)) |
| { |
| // is this row for a new time slice? |
| if (sliceNum > currSliceNum_) |
| { |
| currSliceNum_ = sliceNum; |
| entriesForThisTimeSlice_ = 0; |
| } |
| |
| // we keep up to 2 entries for each time slice, the |
| // first value we see and the last value |
| if (entriesForThisTimeSlice_ < 2) |
| { |
| // first or second value, extend the vectors |
| if (useLong_) |
| lValues_.push_back(NullableTimedValue<long>(wasNull, t, lVal)); |
| else |
| dValues_.push_back(NullableTimedValue<double>(wasNull, t, dVal)); |
| currIx_++; |
| entriesForThisTimeSlice_++; |
| } |
| else |
| { |
| // subsequent value, overwrite the last value seen, we don't need |
| // to keep values that are neither the first nor the last value in |
| // this time slice |
| if (useLong_) |
| lValues_[currIx_] = NullableTimedValue<long>(wasNull, t, lVal); |
| else |
| dValues_[currIx_] = NullableTimedValue<double>(wasNull, t, dVal); |
| } |
| } |
| } |
| |
| void TimeSeriesAggregate::finalizePartition() |
| { |
| // print out the gathered data for debugging |
| /* |
| printf("Aggregate column # %d\n", outputColNum_); |
| for (int i=0; i<numEntries(); i++) |
| { |
| printf("Time: %d, value: ", static_cast<int>(getTime(i))); |
| if (useLong_) |
| { |
| if (lValues_[i].isNull()) |
| printf("NULL\n"); |
| else |
| printf("%ld\n", lValues_[i].getVal()); |
| } |
| else |
| { |
| if (dValues_[i].isNull()) |
| printf("NULL\n"); |
| else |
| printf("%g\n", dValues_[i].getVal()); |
| } |
| } |
| */ |
| currIx_ = -1; |
| currSliceNum_ = entriesForThisTimeSlice_ = -1; |
| } |
| |
| void TimeSeriesAggregate::setOutputCol(time_t startTime, |
| int sliceNum, |
| time_t width) |
| { |
| time_t targetTime = (isFirstVal_ ? startTime : startTime + width); |
| time_t currTime = 0; |
| |
| // Find the last entry before our target time, if it exists. |
| // Include the target time for first value, exclude it for |
| // last value (in that case the target time already belongs |
| // to the next time slice) |
| while (currIx_+1 < numEntries() && |
| ((isFirstVal_ && getTime(currIx_+1) <= targetTime) || |
| (!isFirstVal_ && getTime(currIx_+1) < targetTime))) |
| currIx_++; |
| |
| if (currIx_ >= 0) |
| currTime = getTime(currIx_); |
| |
| if (isConstInterpol_) |
| { |
| // currIx_ points to the entry with the last known constant |
| // value to use or currIx_ is -1 (no last known value exists) |
| if (currIx_ < 0 || |
| (useLong_ && lValues_[currIx_].isNull()) || |
| (!useLong_ && dValues_[currIx_].isNull())) |
| // produce a NULL value |
| outTup_.setNull(outputColNum_); |
| else if (useLong_) |
| outTup_.setLong(outputColNum_, lValues_[currIx_].getVal()); |
| else |
| outTup_.setDouble(outputColNum_, dValues_[currIx_].getVal()); |
| } // constant interpolation |
| else |
| { |
| // linear interpolation, always uses a double value |
| |
| // currIx_ points to the last known value or is -1 (no last |
| // known value exists). Now look for the next known value. |
| NullableTimedValue<double> interpolatedVal; |
| |
| if (currIx_ >= 0 && currTime == targetTime) |
| { |
| // currIx_ and nextIx_ point to the one |
| // entry describing the value, no interpolation needed |
| interpolatedVal = dValues_[currIx_]; |
| } |
| else if (currIx_ < 0 || currIx_+1 >= numEntries()) |
| { |
| // produce a NULL value for these boundary cases where |
| // we are missing an upper or lower entry to use for |
| // interpolation |
| interpolatedVal = NullableTimedValue<double>(true, targetTime, 0.0); |
| } |
| else |
| { |
| // currIx_ and the next entry should have time values |
| // that form an interval we can use to interpolate |
| interpolatedVal = |
| dValues_[currIx_].interpolateLinear( |
| targetTime, |
| dValues_[currIx_+1]); |
| } |
| |
| // now produce the value |
| if (interpolatedVal.isNull()) |
| outTup_.setNull(outputColNum_); |
| else |
| outTup_.setDouble(outputColNum_, interpolatedVal.getVal()); |
| } // linear interpolation |
| } |
| |
| |
| |
| InternalColumns::InternalColumns(const UDRInvocationInfo &info) : |
| info_(info) |
| { |
| // expect a single table-valued input |
| if (info.getNumTableInputs() != 1) |
| throw UDRException( |
| 38010, |
| "TIMESERIES UDF: Expecting one table-valued input"); |
| |
| const OrderInfo &ord = info.in().getQueryOrdering(); |
| const PartitionInfo &part = info.in().getQueryPartitioning(); |
| |
| // perform some basic tests in the first call at compile time |
| if (info.getCallPhase() == UDRInvocationInfo::COMPILER_INITIAL_CALL) |
| { |
| // expect an order by on a time or timestamp expression |
| if (ord.getNumEntries() != 1 || |
| ord.getOrderType(0) == OrderInfo::DESCENDING) |
| throw UDRException( |
| 38020, |
| "TIMESERIES UDF: Must use ORDER BY with one column for its input table and the order must be ascending"); |
| TypeInfo::SQLTypeCode typeCode = |
| info.in().getColumn(ord.getColumnNum(0)).getType().getSQLType(); |
| if (typeCode != TypeInfo::TIME && |
| typeCode != TypeInfo::TIMESTAMP) |
| throw UDRException( |
| 38030, |
| "TIMESERIES UDF: Must use ORDER BY a TIME or TIMESTAMP column for the input table"); |
| |
| // we need at least two parameters, time column name and width |
| // of time slice |
| if (info.par().getNumColumns() < 2) |
| throw UDRException( |
| 38040, |
| "TIMESERIES UDF: UDF needs to be called with at least 2 scalar parameters"); |
| |
| // input parameter 0 (defined in the DDL) is the |
| // name of the column containing the time values |
| if (!info.par().isAvailable(0) || |
| info.par().getColumn(0).getType().getSQLTypeClass() != |
| TypeInfo::CHARACTER_TYPE) |
| throw UDRException( |
| 38050, |
| "TIMESERIES UDF: Expecting a character constant (timestamp alias) as first parameter"); |
| |
| // check type and value of the time slice width, specified |
| // as parameter 1 |
| if (!info.par().isAvailable(1)) |
| throw UDRException( |
| 38060, |
| "TIMESERIES UDF: Expecting a constant for the time slice width as second parameter"); |
| |
| // time slice width must be a day-second interval |
| if (info.par().getColumn(1).getType().getSQLTypeSubClass() != |
| TypeInfo::DAY_SECOND_INTERVAL_TYPE) |
| throw UDRException( |
| 38070, |
| "TIMESERIES UDF: Second scalar parameter for time slice width must be an interval constant in the day to second range"); |
| |
| // make sure parameters come in pairs |
| if (info.par().getNumColumns() % 2 != 0) |
| throw UDRException( |
| 38080, |
| "TIMESERIES UDF: Parameters need to be specified in pairs of column name and instructions"); |
| |
| // make sure all parameters are specified at compile time |
| for (int p=2; p<info.par().getNumColumns(); p++) |
| if (!info.par().isAvailable(p)) |
| throw UDRException( |
| 38090, |
| "TIMESERIES UDF: All parameters must be specified as literals"); |
| } // initial compile-time checks |
| |
| tsInColNum_ = ord.getColumnNum(0); |
| numTSCols_ = 1; // always a single timestamp column for now |
| numPartCols_ = part.getNumEntries(); |
| |
| timeSliceWidth_ = info.par().getTime(1); |
| |
| // initialize vectors |
| for (int p=0; p<numPartCols_; p++) |
| { |
| currPartKey_.push_back(""); |
| currPartKeyNulls_.push_back(true); |
| } |
| |
| int ip = 2; |
| |
| while (ip<info.par().getNumColumns()) |
| { |
| std::string colName = info.par().getString(ip); |
| std::string instr = info.par().getString(ip+1); |
| bool isFirstVal; |
| bool isConstInterpol; |
| bool isIgnoreNulls; |
| |
| // some checks done only during the first compile time call |
| if (info.getCallPhase() == UDRInvocationInfo::COMPILER_INITIAL_CALL) |
| { |
| if (instr.size() < 2 || instr.size() > 3) |
| throw UDRException( |
| 38100, |
| "TIMESERIES UDF: Expecting instructions with 2 or 3 characters: %s", |
| instr.c_str()); |
| |
| // validate first character of instructions |
| switch (instr[0]) |
| { |
| case 'f': |
| case 'F': |
| case 'l': |
| case 'L': |
| break; |
| |
| default: |
| throw UDRException( |
| 38110, |
| "TIMESERIES UDF: Parameter %d should start with F or L for first or last value", |
| ip+2); |
| } |
| |
| // validate second character of instructions |
| switch (instr[1]) |
| { |
| case 'c': |
| case 'C': |
| case 'l': |
| case 'L': |
| break; |
| |
| default: |
| throw UDRException( |
| 38120, |
| "TIMESERIES UDF: Parameter %d should have C or L as its second character, for constant or linear interpolation", |
| ip+2); |
| } |
| |
| if (instr.size() == 3 && |
| instr[2] != 'i' && |
| instr[2] != 'I') |
| throw UDRException( |
| 38130, |
| "TIMESERIES UDF: Unexpected trailing characters in aggregate instructions: %s", |
| instr.c_str()); |
| } // compile-time checks |
| |
| isFirstVal = (instr[0] == 'F' || instr[0] == 'f'); |
| isConstInterpol = (instr[1] == 'C' || instr[1] == 'c'); |
| isIgnoreNulls = (instr.size() > 2); |
| |
| columns_.push_back(new TimeSeriesAggregate( |
| info.in(), |
| info.out(), |
| info.in().getColNum(info.par().getString(ip)), |
| getFirstAggrCol() + columns_.size(), |
| isFirstVal, |
| isConstInterpol, |
| isIgnoreNulls)); |
| ip += 2; |
| } |
| } |
| |
| InternalColumns::~InternalColumns() |
| { |
| for (std::vector<TimeSeriesAggregate *>::iterator it = columns_.begin(); |
| it != columns_.end(); |
| it++) |
| delete *it; |
| } |
| |
| void InternalColumns::initializePartition() |
| { |
| const PartitionInfo &pi = info_.in().getQueryPartitioning(); |
| |
| // get first value of the time stamp column and round it down |
| // to the next multiple of the time stamp width |
| partitionStartTime_ = info_.in().getTime(tsInColNum_); |
| partitionStartTime_ -= partitionStartTime_ % timeSliceWidth_; |
| |
| // start and end time are the same for the first row |
| partitionCurrTime_ = partitionStartTime_; |
| |
| // remember the partition key values as an array of strings |
| for (int p=0; p<numPartCols_; p++) |
| { |
| currPartKey_[p] = info_.in().getString(pi.getColumnNum(p)); |
| currPartKeyNulls_[p] = info_.in().wasNull(); |
| } |
| |
| for (std::vector<TimeSeriesAggregate *>::iterator it = columns_.begin(); |
| it != columns_.end(); |
| it++) |
| (*it)->initPartition(); |
| } |
| |
| bool InternalColumns::isSamePartition() |
| { |
| const PartitionInfo &pi = info_.in().getQueryPartitioning(); |
| std::string pkVal; |
| bool wasNull; |
| |
| // compare partitioning keys |
| for (int p=0; p<numPartCols_; p++) |
| { |
| pkVal = info_.in().getString(pi.getColumnNum(p)); |
| wasNull = info_.in().wasNull(); |
| |
| if (wasNull != currPartKeyNulls_[p] || |
| (!wasNull && pkVal != currPartKey_[p])) |
| return false; |
| } |
| |
| // partition key matched, we are still in the same partition |
| return true; |
| } |
| |
| int InternalColumns::getNumTimeSlices() |
| { |
| return (partitionCurrTime_ - partitionStartTime_) / timeSliceWidth_ + 1; |
| } |
| |
| void InternalColumns::readInputCols() |
| { |
| int sliceNum; |
| |
| // remember the highest time value we have seen for this partition |
| partitionCurrTime_ = info_.in().getTime(tsInColNum_); |
| sliceNum = getNumTimeSlices() - 1; |
| |
| for (std::vector<TimeSeriesAggregate *>::iterator it = columns_.begin(); |
| it != columns_.end(); |
| it++) |
| (*it)->readInputCol(partitionCurrTime_, sliceNum); |
| } |
| |
| void InternalColumns::finalizePartition() |
| { |
| for (std::vector<TimeSeriesAggregate *>::iterator it = columns_.begin(); |
| it != columns_.end(); |
| it++) |
| (*it)->finalizePartition(); |
| } |
| |
| void InternalColumns::setOutputCols(int sliceNum) |
| { |
| time_t timeSliceStartTime = partitionStartTime_ + sliceNum * timeSliceWidth_; |
| |
| // set the partitioning columns |
| for (int p=0; p<numPartCols_; p++) |
| { |
| if (currPartKeyNulls_[p]) |
| info_.out().setNull(p); |
| else |
| info_.out().setString(p,currPartKey_[p]); |
| } |
| |
| // set the time slice column |
| info_.out().setTime( |
| numPartCols_, |
| timeSliceStartTime); |
| |
| // set the aggregate columns |
| for (std::vector<TimeSeriesAggregate *>::iterator it = columns_.begin(); |
| it != columns_.end(); |
| it++) |
| (*it)->setOutputCol( |
| timeSliceStartTime, |
| sliceNum, |
| timeSliceWidth_); |
| } |
| |
