QRunIO/qqq
2
0
Fork 0
mirror of https://github.com/Kingsrook/qqq.git synced 2025-07-18 05:01:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

QQQ-37 update streamed-etl steps to not have to use different record-list

Browse Source
This commit is contained in:
darin.kelkhoff
2022-09-06 15:15:04 -05:00
parent 31e6bf4d49
commit 25c9376ce4
13 changed files with 229 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/model/actions/processes/RunBackendStepInput.java
View File

@ -51,6 +51,10 @@ public class RunBackendStepInput extends AbstractActionInput
private QProcessCallback callback;
private AsyncJobCallback asyncJobCallback;
////////////////////////////////////////////////////////////////////////////
// note - new fields should generally be added in method: cloneFieldsInto //
////////////////////////////////////////////////////////////////////////////
/*******************************************************************************
@ -85,6 +89,25 @@ public class RunBackendStepInput extends AbstractActionInput
/*******************************************************************************
** Kinda like a reverse copy-constructor -- for a subclass that wants all the
** field values from this object. Keep this in sync with the fields in this class!
**
** Of note - the processState does NOT get cloned - because... well, in our first
** use-case (a subclass that doesn't WANT the same/full state), that's what we needed.
*******************************************************************************/
public void cloneFieldsInto(RunBackendStepInput target)
{
target.setStepName(getStepName());
target.setSession(getSession());
target.setTableName(getTableName());
target.setProcessName(getProcessName());
target.setAsyncJobCallback(getAsyncJobCallback());
target.setValues(getValues());
}
/*******************************************************************************
**
*******************************************************************************/
@ -429,4 +452,5 @@ public class RunBackendStepInput extends AbstractActionInput
}
return (asyncJobCallback);
}
}

58
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamedwithfrontend/AbstractLoadStep.java
View File

@ -22,31 +22,27 @@
package com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
import com.kingsrook.qqq.backend.core.actions.processes.BackendStep;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
/*******************************************************************************
** Base class for the Load (aka, store) logic of Streamed ETL processes.
**
** Records are to be read out of the inputRecordPage field, and after storing,
** should be written to the outputRecordPage. That is to say, DO NOT use the
** recordList in the step input/output objects.
** Records are to be read out of the input object's Records field, and after storing,
** should be written to the output object's Records, noting that when running
** as a streamed-ETL process, those input & output objects will be instances of
** the StreamedBackendStep{Input,Output} classes, that will be associated with
** a page of records flowing thorugh a pipe.
**
** Also - use the transaction member variable!!!
*******************************************************************************/
public abstract class AbstractLoadStep implements BackendStep
{
private List<QRecord> inputRecordPage = new ArrayList<>();
private List<QRecord> outputRecordPage = new ArrayList<>();
private Optional<QBackendTransaction> transaction = Optional.empty();
@ -87,50 +83,6 @@ public abstract class AbstractLoadStep implements BackendStep
/*******************************************************************************
** Getter for recordPage
**
*******************************************************************************/
public List<QRecord> getInputRecordPage()
{
return inputRecordPage;
}
/*******************************************************************************
** Setter for recordPage
**
*******************************************************************************/
public void setInputRecordPage(List<QRecord> inputRecordPage)
{
this.inputRecordPage = inputRecordPage;
}
/*******************************************************************************
** Getter for outputRecordPage
**
*******************************************************************************/
public List<QRecord> getOutputRecordPage()
{
return outputRecordPage;
}
/*******************************************************************************
** Setter for outputRecordPage
**
*******************************************************************************/
public void setOutputRecordPage(List<QRecord> outputRecordPage)
{
this.outputRecordPage = outputRecordPage;
}
/*******************************************************************************
** Setter for transaction
**

58
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamedwithfrontend/AbstractTransformStep.java
View File

@ -22,27 +22,24 @@
package com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend;
import java.util.ArrayList;
import java.util.List;
import com.kingsrook.qqq.backend.core.actions.processes.BackendStep;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
/*******************************************************************************
** Base class for the Transform logic of Streamed ETL processes.
**
** Records are to be read out of the inputRecordPage field, and after transformation,
** should be written to the outputRecordPage. That is to say, DO NOT use the
** recordList in the step input/output objects.
** Records are to be read out of the input object's Records field, and after storing,
** should be written to the output object's Records, noting that when running
** as a streamed-ETL process, those input & output objects will be instances of
** the StreamedBackendStep{Input,Output} classes, that will be associated with
** a page of records flowing through a pipe.
**
*******************************************************************************/
public abstract class AbstractTransformStep implements BackendStep, ProcessSummaryProviderInterface
{
private List<QRecord> inputRecordPage = new ArrayList<>();
private List<QRecord> outputRecordPage = new ArrayList<>();
/*******************************************************************************
@ -69,47 +66,4 @@ public abstract class AbstractTransformStep implements BackendStep, ProcessSumma
////////////////////////
}
/*******************************************************************************
** Getter for recordPage
**
*******************************************************************************/
public List<QRecord> getInputRecordPage()
{
return inputRecordPage;
}
/*******************************************************************************
** Setter for recordPage
**
*******************************************************************************/
public void setInputRecordPage(List<QRecord> inputRecordPage)
{
this.inputRecordPage = inputRecordPage;
}
/*******************************************************************************
** Getter for outputRecordPage
**
*******************************************************************************/
public List<QRecord> getOutputRecordPage()
{
return outputRecordPage;
}
/*******************************************************************************
** Setter for outputRecordPage
**
*******************************************************************************/
public void setOutputRecordPage(List<QRecord> outputRecordPage)
{
this.outputRecordPage = outputRecordPage;
}
}

4
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamedwithfrontend/LoadViaInsertStep.java
View File

@ -52,10 +52,10 @@ public class LoadViaInsertStep extends AbstractLoadStep
InsertInput insertInput = new InsertInput(runBackendStepInput.getInstance());
insertInput.setSession(runBackendStepInput.getSession());
insertInput.setTableName(runBackendStepInput.getValueString(FIELD_DESTINATION_TABLE));
insertInput.setRecords(getInputRecordPage());
insertInput.setRecords(runBackendStepInput.getRecords());
getTransaction().ifPresent(insertInput::setTransaction);
InsertOutput insertOutput = new InsertAction().execute(insertInput);
getOutputRecordPage().addAll(insertOutput.getRecords());
runBackendStepOutput.getRecords().addAll(insertOutput.getRecords());
}

4
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamedwithfrontend/LoadViaUpdateStep.java
View File

@ -54,10 +54,10 @@ public class LoadViaUpdateStep extends AbstractLoadStep
UpdateInput updateInput = new UpdateInput(runBackendStepInput.getInstance());
updateInput.setSession(runBackendStepInput.getSession());
updateInput.setTableName(runBackendStepInput.getValueString(FIELD_DESTINATION_TABLE));
updateInput.setRecords(getInputRecordPage());
updateInput.setRecords(runBackendStepInput.getRecords());
getTransaction().ifPresent(updateInput::setTransaction);
UpdateOutput updateOutput = new UpdateAction().execute(updateInput);
getOutputRecordPage().addAll(updateOutput.getRecords());
runBackendStepOutput.getRecords().addAll(updateOutput.getRecords());
}

77
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamedwithfrontend/StreamedBackendStepInput.java Normal file
View File

@ -0,0 +1,77 @@
/*
* QQQ - Low-code Application Framework for Engineers.
* Copyright (C) 2021-2022. Kingsrook, LLC
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
* contact@kingsrook.com
* https://github.com/Kingsrook/
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend;
import java.util.List;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
/*******************************************************************************
** Subclass of RunBackendStepInput, meant for use in the pseudo-steps used by
** the Streamed-ETL-with-frontend processes - where the Record list is not the
** full process's record list - rather - is just a page at a time -- so this class
** overrides the getRecords and setRecords method, to just work with that page.
**
** Note - of importance over time may be the RunBackendStepInput::cloneFieldsInto
** method - e.g., if new fields are added to that class!
*******************************************************************************/
public class StreamedBackendStepInput extends RunBackendStepInput
{
private List<QRecord> inputRecords;
/*******************************************************************************
**
*******************************************************************************/
public StreamedBackendStepInput(RunBackendStepInput runBackendStepInput, List<QRecord> inputRecords)
{
super(runBackendStepInput.getInstance());
runBackendStepInput.cloneFieldsInto(this);
this.inputRecords = inputRecords;
}
/*******************************************************************************
**
*******************************************************************************/
@Override
public void setRecords(List<QRecord> records)
{
this.inputRecords = records;
}
/*******************************************************************************
**
*******************************************************************************/
@Override
public List<QRecord> getRecords()
{
return (inputRecords);
}
}

74
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamedwithfrontend/StreamedBackendStepOutput.java Normal file
View File

@ -0,0 +1,74 @@
/*
* QQQ - Low-code Application Framework for Engineers.
* Copyright (C) 2021-2022. Kingsrook, LLC
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
* contact@kingsrook.com
* https://github.com/Kingsrook/
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend;
import java.util.ArrayList;
import java.util.List;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
/*******************************************************************************
** Subclass of RunBackendStepOutput, meant for use in the pseudo-steps used by
** the Streamed-ETL-with-frontend processes - where the Record list is not the
** full process's record list - rather - is just a page at a time -- so this class
** overrides the getRecords and setRecords method, to just work with that page.
*******************************************************************************/
public class StreamedBackendStepOutput extends RunBackendStepOutput
{
private List<QRecord> outputRecords = new ArrayList<>();
/*******************************************************************************
**
*******************************************************************************/
public StreamedBackendStepOutput(RunBackendStepOutput runBackendStepOutput)
{
super();
setValues(runBackendStepOutput.getValues());
}
/*******************************************************************************
**
*******************************************************************************/
@Override
public void setRecords(List<QRecord> records)
{
this.outputRecords = records;
}
/*******************************************************************************
**
*******************************************************************************/
@Override
public List<QRecord> getRecords()
{
return (outputRecords);
}
}

21
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamedwithfrontend/StreamedETLExecuteStep.java
View File

@ -144,27 +144,32 @@ public class StreamedETLExecuteStep extends BaseStreamedETLStep implements Backe
///////////////////////////////////
List<QRecord> qRecords = recordPipe.consumeAvailableRecords();
///////////////////////////////////////////////////////////////////////
// make streamed input & output objects from the run input & outputs //
///////////////////////////////////////////////////////////////////////
StreamedBackendStepInput streamedBackendStepInput = new StreamedBackendStepInput(runBackendStepInput, qRecords);
StreamedBackendStepOutput streamedBackendStepOutput = new StreamedBackendStepOutput(runBackendStepOutput);
/////////////////////////////////////////////////////
// pass the records through the transform function //
/////////////////////////////////////////////////////
transformStep.setInputRecordPage(qRecords);
transformStep.setOutputRecordPage(new ArrayList<>());
transformStep.run(runBackendStepInput, runBackendStepOutput);
transformStep.run(streamedBackendStepInput, streamedBackendStepOutput);
////////////////////////////////////////////////
// pass the records through the load function //
////////////////////////////////////////////////
loadStep.setInputRecordPage(transformStep.getOutputRecordPage());
loadStep.setOutputRecordPage(new ArrayList<>());
loadStep.run(runBackendStepInput, runBackendStepOutput);
streamedBackendStepInput = new StreamedBackendStepInput(runBackendStepInput, streamedBackendStepOutput.getRecords());
streamedBackendStepOutput = new StreamedBackendStepOutput(runBackendStepOutput);
loadStep.run(streamedBackendStepInput, streamedBackendStepOutput);
///////////////////////////////////////////////////////
// copy a small number of records to the output list //
///////////////////////////////////////////////////////
int i = 0;
while(loadedRecordList.size() < PROCESS_OUTPUT_RECORD_LIST_LIMIT && i < loadStep.getOutputRecordPage().size())
while(loadedRecordList.size() < PROCESS_OUTPUT_RECORD_LIST_LIMIT && i < streamedBackendStepOutput.getRecords().size())
{
loadedRecordList.add(loadStep.getOutputRecordPage().get(i++));
loadedRecordList.add(streamedBackendStepOutput.getRecords().get(i++));
}
currentRowCount += qRecords.size();

11
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamedwithfrontend/StreamedETLPreviewStep.java
View File

@ -151,16 +151,21 @@ public class StreamedETLPreviewStep extends BaseStreamedETLStep implements Backe
///////////////////////////////////
List<QRecord> qRecords = recordPipe.consumeAvailableRecords();
///////////////////////////////////////////////////////////////////////
// make streamed input & output objects from the run input & outputs //
///////////////////////////////////////////////////////////////////////
StreamedBackendStepInput streamedBackendStepInput = new StreamedBackendStepInput(runBackendStepInput, qRecords);
StreamedBackendStepOutput streamedBackendStepOutput = new StreamedBackendStepOutput(runBackendStepOutput);
/////////////////////////////////////////////////////
// pass the records through the transform function //
/////////////////////////////////////////////////////
transformStep.setInputRecordPage(qRecords);
transformStep.run(runBackendStepInput, runBackendStepOutput);
transformStep.run(streamedBackendStepInput, streamedBackendStepOutput);
////////////////////////////////////////////////////
// add the transformed records to the output list //
////////////////////////////////////////////////////
transformedRecordList.addAll(transformStep.getOutputRecordPage());
transformedRecordList.addAll(streamedBackendStepOutput.getRecords());
return (qRecords.size());
}

13
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamedwithfrontend/StreamedETLValidateStep.java
View File

@ -128,19 +128,24 @@ public class StreamedETLValidateStep extends BaseStreamedETLStep implements Back
///////////////////////////////////
List<QRecord> qRecords = recordPipe.consumeAvailableRecords();
///////////////////////////////////////////////////////////////////////
// make streamed input & output objects from the run input & outputs //
///////////////////////////////////////////////////////////////////////
StreamedBackendStepInput streamedBackendStepInput = new StreamedBackendStepInput(runBackendStepInput, qRecords);
StreamedBackendStepOutput streamedBackendStepOutput = new StreamedBackendStepOutput(runBackendStepOutput);
/////////////////////////////////////////////////////
// pass the records through the transform function //
/////////////////////////////////////////////////////
transformStep.setInputRecordPage(qRecords);
transformStep.run(runBackendStepInput, runBackendStepOutput);
transformStep.run(streamedBackendStepInput, streamedBackendStepOutput);
///////////////////////////////////////////////////////
// copy a small number of records to the output list //
///////////////////////////////////////////////////////
int i = 0;
while(previewRecordList.size() < PROCESS_OUTPUT_RECORD_LIST_LIMIT && i < transformStep.getOutputRecordPage().size())
while(previewRecordList.size() < PROCESS_OUTPUT_RECORD_LIST_LIMIT && i < streamedBackendStepOutput.getRecords().size())
{
previewRecordList.add(transformStep.getOutputRecordPage().get(i++));
previewRecordList.add(streamedBackendStepOutput.getRecords().get(i++));
}
currentRowCount += qRecords.size();

12
qqq-backend-core/src/test/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamedwithfrontend/StreamedETLWithFrontendProcessTest.java
View File

@ -347,12 +347,12 @@ public class StreamedETLWithFrontendProcessTest
@Override
public void run(RunBackendStepInput runBackendStepInput, RunBackendStepOutput runBackendStepOutput) throws QException
{
for(QRecord qRecord : getInputRecordPage())
for(QRecord qRecord : runBackendStepInput.getRecords())
{
QRecord newQRecord = new QRecord();
newQRecord.setValue("firstName", "Johnny");
newQRecord.setValue("lastName", qRecord.getValueString("name"));
getOutputRecordPage().add(newQRecord);
runBackendStepOutput.getRecords().add(newQRecord);
}
}
@ -403,7 +403,7 @@ public class StreamedETLWithFrontendProcessTest
@Override
public void run(RunBackendStepInput runBackendStepInput, RunBackendStepOutput runBackendStepOutput) throws QException
{
for(QRecord qRecord : getInputRecordPage())
for(QRecord qRecord : runBackendStepInput.getRecords())
{
if(qRecord.getValueString("name").equals("Circle"))
{
@ -414,7 +414,7 @@ public class StreamedETLWithFrontendProcessTest
QRecord newQRecord = new QRecord();
newQRecord.setValue("firstName", "Johnny");
newQRecord.setValue("lastName", qRecord.getValueString("name"));
getOutputRecordPage().add(newQRecord);
runBackendStepOutput.getRecords().add(newQRecord);
okSummary.incrementCountAndAddPrimaryKey(qRecord.getValue("id"));
}
@ -437,12 +437,12 @@ public class StreamedETLWithFrontendProcessTest
@Override
public void run(RunBackendStepInput runBackendStepInput, RunBackendStepOutput runBackendStepOutput) throws QException
{
for(QRecord qRecord : getInputRecordPage())
for(QRecord qRecord : runBackendStepInput.getRecords())
{
QRecord updatedQRecord = new QRecord();
updatedQRecord.setValue("id", qRecord.getValue("id"));
updatedQRecord.setValue("name", "Transformed:" + qRecord.getValueString("name"));
getOutputRecordPage().add(updatedQRecord);
runBackendStepOutput.getRecords().add(updatedQRecord);
}
}