add cache support for insert
@@ -31,6 +31,7 @@ import io.siddhi.core.event.stream.StreamEvent;
import io.siddhi.core.event.stream.StreamEventCloner;
import io.siddhi.core.exception.ConnectionUnavailableException;
import io.siddhi.core.exception.SiddhiAppCreationException;
+import io.siddhi.core.exception.SiddhiAppRuntimeException;
import io.siddhi.core.executor.ConstantExpressionExecutor;
import io.siddhi.core.executor.ExpressionExecutor;
import io.siddhi.core.executor.VariableExpressionExecutor;
@@ -151,6 +152,7 @@ public abstract class AbstractQueryableRecordTable extends AbstractRecordTable i
connect();
if (isCacheEnabled) {
+ try {
StateEvent stateEvent = new StateEvent(1, 0);
StreamEvent preLoadedData = query(stateEvent, compiledConditionForCaching, compiledSelectionForCaching,
outputAttributesForCaching);
@@ -174,6 +176,9 @@ public abstract class AbstractQueryableRecordTable extends AbstractRecordTable i
log.warn("Table size bigger than cache table size. So cache is left empty");
}
}
+ } catch (SiddhiAppRuntimeException ignore) {
+
+ }
}
}
@@ -228,6 +233,26 @@ public abstract class AbstractQueryableRecordTable extends AbstractRecordTable i
}
+ @Override
+ public void add(ComplexEventChunk<StreamEvent> addingEventChunk) throws ConnectionUnavailableException {
+ if (isCacheEnabled) {
+ cachedTable.add(addingEventChunk);
+ }
+ List<Object[]> records = new ArrayList<>();
+ addingEventChunk.reset();
+ long timestamp = 0L;
+ while (addingEventChunk.hasNext()) {
+ StreamEvent event = addingEventChunk.next();
+ records.add(event.getOutputData());
+ timestamp = event.getTimestamp();
+ }
+ if (recordTableHandler != null) {
+ recordTableHandler.add(timestamp, records);
+ } else {
+ add(records);
+ }
+ }
+
@Override
public CompiledCondition compileCondition(Expression condition, MatchingMetaInfoHolder matchingMetaInfoHolder,
List<VariableExpressionExecutor> variableExpressionExecutors,
