PipelineExecutor: wrong groupBuffer initialization
@@ -104,7 +104,6 @@ public class PipelineExecutor extends BaseTransform<PipelineExecutorMeta, Pipeli
logDetailed("Identified a new pipeline to execute: '" + filename + "'");
meta.setFilename(filename);
pipelineExecutorData.prevFilename = filename;
- pipelineExecutorData.groupBuffer = null;
initPipeline(pipelineExecutorData);
initOnFirstProcessingIteration();
}
@@ -502,6 +501,29 @@ public class PipelineExecutor extends BaseTransform<PipelineExecutorMeta, Pipeli
boolean transformSuccessfullyInitialized = super.init();
if (transformSuccessfullyInitialized) {
+
+ // Init subpipeline commons parameters and data structures
+ // We are sure groupBuffer is initialized only once
+ pipelineExecutorData.groupBuffer = new ArrayList<>();
+
+ // How many rows do we group together for the pipeline?
+ if (!Utils.isEmpty(meta.getGroupSize())) {
+ pipelineExecutorData.groupSize = Const.toInt(resolve(meta.getGroupSize()), -1);
+ } else {
+ pipelineExecutorData.groupSize = -1;
+ }
+ // Is there a grouping time set?
+ if (!Utils.isEmpty(meta.getGroupTime())) {
+ pipelineExecutorData.groupTime = Const.toInt(resolve(meta.getGroupTime()), -1);
+ } else {
+ pipelineExecutorData.groupTime = -1;
+ }
+
+ // Is there a grouping field set?
+ if (!Utils.isEmpty(meta.getGroupField())) {
+ pipelineExecutorData.groupField = resolve(meta.getGroupField());
+ }
+
// First we need to load the mapping (pipeline)
try {
if ((!meta.isFilenameInField() && Utils.isEmpty(meta.getFilename()))
@@ -530,26 +552,8 @@ public class PipelineExecutor extends BaseTransform<PipelineExecutorMeta, Pipeli
// Do we have a pipeline at all?
if (pipelineExecutorData.getExecutorPipelineMeta() != null) {
- pipelineExecutorData.groupBuffer = new ArrayList<>();
-
- // How many rows do we group together for the pipeline?
- if (!Utils.isEmpty(meta.getGroupSize())) {
- pipelineExecutorData.groupSize = Const.toInt(resolve(meta.getGroupSize()), -1);
- } else {
- pipelineExecutorData.groupSize = -1;
- }
- // Is there a grouping time set?
- if (!Utils.isEmpty(meta.getGroupTime())) {
- pipelineExecutorData.groupTime = Const.toInt(resolve(meta.getGroupTime()), -1);
- } else {
- pipelineExecutorData.groupTime = -1;
- }
+ // Init timing for group time timing
pipelineExecutorData.groupTimeStart = System.currentTimeMillis();
-
- // Is there a grouping field set?
- if (!Utils.isEmpty(meta.getGroupField())) {
- pipelineExecutorData.groupField = resolve(meta.getGroupField());
- }
// That's all for now...
return true;
} else {
