Changes pushed to qqq-backend-module-filesystem (solo-repo) in 0.2 support

qqq-backend-module-filesystem/src/main/java/com/kingsrook/qqq/backend/module/filesystem/base/actions/AbstractBaseFilesystemAction.java

@@ -201,10 +201,16 @@ public abstract class AbstractBaseFilesystemAction<FILE>
                   String fileContents = IOUtils.toString(readFile(file));
                   fileContents = customizeFileContentsAfterReading(table, fileContents);
 
+                  if(queryInput.getRecordPipe() != null)
+                  {
+                     new CsvToQRecordAdapter().buildRecordsFromCsv(queryInput.getRecordPipe(), fileContents, table, null, (record -> addBackendDetailsToRecord(record, file)));
+                  }
+                  else
+                  {
                      List<QRecord> recordsInFile = new CsvToQRecordAdapter().buildRecordsFromCsv(fileContents, table, null);
                      addBackendDetailsToRecords(recordsInFile, file);
-
                      queryOutput.addRecords(recordsInFile);
+                  }
                   break;
                }
                case JSON:
@@ -212,6 +218,7 @@ public abstract class AbstractBaseFilesystemAction<FILE>
                   String fileContents = IOUtils.toString(readFile(file));
                   fileContents = customizeFileContentsAfterReading(table, fileContents);
 
+                  // todo - pipe support!!
                   List<QRecord> recordsInFile = new JsonToQRecordAdapter().buildRecordsFromJson(fileContents, table, null);
                   addBackendDetailsToRecords(recordsInFile, file);
 
@@ -241,10 +248,17 @@ public abstract class AbstractBaseFilesystemAction<FILE>
     *******************************************************************************/
    protected void addBackendDetailsToRecords(List<QRecord> recordsInFile, FILE file)
    {
-      recordsInFile.forEach(record ->
+      recordsInFile.forEach(r -> addBackendDetailsToRecord(r, file));
+   }
+
+
+
+   /*******************************************************************************
+    ** Add backend details to a record about the file that it is in.
+    *******************************************************************************/
+   protected void addBackendDetailsToRecord(QRecord record, FILE file)
    {
-         record.withBackendDetail(FilesystemRecordBackendDetailFields.FULL_PATH, getFullPathForFile(file));
+      record.addBackendDetail(FilesystemRecordBackendDetailFields.FULL_PATH, getFullPathForFile(file));
-      });
    }
 
 

qqq-backend-module-filesystem/src/main/java/com/kingsrook/qqq/backend/module/filesystem/processes/implementations/etl/basic/BasicETLCleanupSourceFilesStep.java

@@ -94,11 +94,13 @@ public class BasicETLCleanupSourceFilesStep implements BackendStep
          String moveOrDelete = runBackendStepInput.getValueString(FIELD_MOVE_OR_DELETE);
          if(VALUE_DELETE.equals(moveOrDelete))
          {
+            LOG.info("Deleting ETL source file: " + sourceFile);
             actionBase.deleteFile(runBackendStepInput.getInstance(), table, sourceFile);
          }
          else if(VALUE_MOVE.equals(moveOrDelete))
          {
             String destinationForMoves = runBackendStepInput.getValueString(FIELD_DESTINATION_FOR_MOVES);
+            LOG.info("Moving ETL source file: " + sourceFile + " to " + destinationForMoves);
             if(!StringUtils.hasContent(destinationForMoves))
             {
                throw (new QException("Field [" + FIELD_DESTINATION_FOR_MOVES + "] is missing a value."));

qqq-backend-module-filesystem/src/main/java/com/kingsrook/qqq/backend/module/filesystem/processes/implementations/etl/streamed/StreamedETLFilesystemBackendStep.java

@@ -0,0 +1,75 @@
+/*
+ * 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.module.filesystem.processes.implementations.etl.streamed;
+
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+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;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamed.StreamedETLBackendStep;
+import com.kingsrook.qqq.backend.core.utils.StringUtils;
+import com.kingsrook.qqq.backend.module.filesystem.base.FilesystemRecordBackendDetailFields;
+import com.kingsrook.qqq.backend.module.filesystem.processes.implementations.etl.basic.BasicETLCollectSourceFileNamesStep;
+
+
+/*******************************************************************************
+ ** Extension to the base StreamedETLBackendStep, unique for where the source
+ ** table is a filesystem, where we want/need to collect the filenames that were
+ ** processed in the Extract step, so they can be passed into the cleanup step.
+ **
+ ** Similar in purpose to the BasicETLCollectSourceFileNamesStep - only in this
+ ** case, due to the streaming behavior of the StreamedETLProcess, we can't really
+ ** inject this code as a separate backend step - so instead we extend that step,
+ ** and override its postTransform method to intercept the records & file names.
+ *******************************************************************************/
+public class StreamedETLFilesystemBackendStep extends StreamedETLBackendStep
+{
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   @Override
+   protected void preTransform(List<QRecord> qRecords, RunBackendStepInput runBackendStepInput, RunBackendStepOutput runBackendStepOutput)
+   {
+      Set<String> sourceFiles = qRecords.stream()
+         .map(record -> record.getBackendDetailString(FilesystemRecordBackendDetailFields.FULL_PATH))
+         .collect(Collectors.toSet());
+
+      /////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+      // expect that we'll be called on multiple "pages" of records as they run through the pipe.                                        //
+      // each time we're called, we need to:                                                                                             //
+      // - get the unique file paths in this list of records                                                                             //
+      // - if we previously set the list of file names in the output, then split that value up and add those names to the set we see now //
+      // - set the list of name (joined by commas) in the output                                                                         //
+      /////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+      String existingListOfFileNames = runBackendStepOutput.getValueString(BasicETLCollectSourceFileNamesStep.FIELD_SOURCE_FILE_PATHS);
+      if(existingListOfFileNames != null)
+      {
+         sourceFiles.addAll(List.of(existingListOfFileNames.split(",")));
+      }
+      runBackendStepOutput.addValue(BasicETLCollectSourceFileNamesStep.FIELD_SOURCE_FILE_PATHS, StringUtils.join(",", sourceFiles));
+   }
+
+}

qqq-backend-module-filesystem/src/main/java/com/kingsrook/qqq/backend/module/filesystem/s3/utils/S3Utils.java

@@ -137,7 +137,7 @@ public class S3Utils
             ////////////////////////////////////////////////////////////////////////////////
             if(key.endsWith("/"))
             {
-               LOG.debug("Skipping file [{}] because it is a folder", key);
+               // LOG.debug("Skipping file [{}] because it is a folder", key);
                continue;
             }
 
@@ -146,7 +146,7 @@ public class S3Utils
             ///////////////////////////////////////////
             if(!pathMatcher.matches(Path.of(URI.create("file:///" + key))))
             {
-               LOG.debug("Skipping file [{}] that does not match glob [{}]", key, glob);
+               // LOG.debug("Skipping file [{}] that does not match glob [{}]", key, glob);
                continue;
             }
 

qqq-backend-module-filesystem/src/test/java/com/kingsrook/qqq/backend/module/filesystem/TestUtils.java

@@ -24,20 +24,27 @@ package com.kingsrook.qqq.backend.module.filesystem;
 
 import java.io.File;
 import java.io.IOException;
-import com.kingsrook.qqq.backend.core.exceptions.QInstanceValidationException;
+import java.util.List;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.instances.QInstanceValidator;
 import com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType;
+import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
 import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QBackendStepMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
 import com.kingsrook.qqq.backend.core.model.session.QSession;
 import com.kingsrook.qqq.backend.core.modules.authentication.MockAuthenticationModule;
 import com.kingsrook.qqq.backend.core.modules.authentication.metadata.QAuthenticationMetaData;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamed.StreamedETLProcess;
 import com.kingsrook.qqq.backend.module.filesystem.base.model.metadata.Cardinality;
 import com.kingsrook.qqq.backend.module.filesystem.base.model.metadata.RecordFormat;
 import com.kingsrook.qqq.backend.module.filesystem.local.model.metadata.FilesystemBackendMetaData;
 import com.kingsrook.qqq.backend.module.filesystem.local.model.metadata.FilesystemTableBackendDetails;
+import com.kingsrook.qqq.backend.module.filesystem.processes.implementations.etl.streamed.StreamedETLFilesystemBackendStep;
 import com.kingsrook.qqq.backend.module.filesystem.s3.BaseS3Test;
 import com.kingsrook.qqq.backend.module.filesystem.s3.model.metadata.S3BackendMetaData;
 import com.kingsrook.qqq.backend.module.filesystem.s3.model.metadata.S3TableBackendDetails;
@@ -51,8 +58,14 @@ public class TestUtils
 {
    public static final String BACKEND_NAME_LOCAL_FS = "local-filesystem";
    public static final String BACKEND_NAME_S3       = "s3";
-   public static final String TABLE_NAME_PERSON_LOCAL_FS = "person";
+   public static final String BACKEND_NAME_MOCK     = "mock";
+
+   public static final String TABLE_NAME_PERSON_LOCAL_FS_JSON = "person-local-json";
+   public static final String TABLE_NAME_PERSON_LOCAL_FS_CSV  = "person-local-csv";
    public static final String TABLE_NAME_PERSON_S3            = "person-s3";
+   public static final String TABLE_NAME_PERSON_MOCK          = "person-mock";
+
+   public static final String PROCESS_NAME_STREAMED_ETL = "etl.streamed";
 
    ///////////////////////////////////////////////////////////////////
    // shouldn't be accessed directly, as we append a counter to it. //
@@ -112,14 +125,18 @@ public class TestUtils
    /*******************************************************************************
     **
     *******************************************************************************/
-   public static QInstance defineInstance() throws QInstanceValidationException
+   public static QInstance defineInstance() throws QException
    {
       QInstance qInstance = new QInstance();
       qInstance.setAuthentication(defineAuthentication());
       qInstance.addBackend(defineLocalFilesystemBackend());
       qInstance.addTable(defineLocalFilesystemJSONPersonTable());
+      qInstance.addTable(defineLocalFilesystemCSVPersonTable());
       qInstance.addBackend(defineS3Backend());
       qInstance.addTable(defineS3CSVPersonTable());
+      qInstance.addBackend(defineMockBackend());
+      qInstance.addTable(defineMockPersonTable());
+      qInstance.addProcess(defineStreamedLocalCsvToMockETLProcess());
 
       new QInstanceValidator().validate(qInstance);
 
@@ -159,21 +176,55 @@ public class TestUtils
    public static QTableMetaData defineLocalFilesystemJSONPersonTable()
    {
       return new QTableMetaData()
-         .withName(TABLE_NAME_PERSON_LOCAL_FS)
+         .withName(TABLE_NAME_PERSON_LOCAL_FS_JSON)
          .withLabel("Person")
          .withBackendName(defineLocalFilesystemBackend().getName())
          .withPrimaryKeyField("id")
-         .withField(new QFieldMetaData("id", QFieldType.INTEGER))
+         .withFields(defineCommonPersonTableFields())
-         .withField(new QFieldMetaData("createDate", QFieldType.DATE_TIME).withBackendName("create_date"))
-         .withField(new QFieldMetaData("modifyDate", QFieldType.DATE_TIME).withBackendName("modify_date"))
-         .withField(new QFieldMetaData("firstName", QFieldType.STRING).withBackendName("first_name"))
-         .withField(new QFieldMetaData("lastName", QFieldType.STRING).withBackendName("last_name"))
-         .withField(new QFieldMetaData("birthDate", QFieldType.DATE).withBackendName("birth_date"))
-         .withField(new QFieldMetaData("email", QFieldType.STRING))
          .withBackendDetails(new FilesystemTableBackendDetails()
             .withBasePath("persons")
             .withRecordFormat(RecordFormat.JSON)
             .withCardinality(Cardinality.MANY)
+            .withGlob("*.json")
+         );
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static List<QFieldMetaData> defineCommonPersonTableFields()
+   {
+      return (List.of(
+         new QFieldMetaData("id", QFieldType.INTEGER),
+         new QFieldMetaData("createDate", QFieldType.DATE_TIME).withBackendName("create_date"),
+         new QFieldMetaData("modifyDate", QFieldType.DATE_TIME).withBackendName("modify_date"),
+         new QFieldMetaData("firstName", QFieldType.STRING).withBackendName("first_name"),
+         new QFieldMetaData("lastName", QFieldType.STRING).withBackendName("last_name"),
+         new QFieldMetaData("birthDate", QFieldType.DATE).withBackendName("birth_date"),
+         new QFieldMetaData("email", QFieldType.STRING)
+      ));
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   public static QTableMetaData defineLocalFilesystemCSVPersonTable()
+   {
+      return new QTableMetaData()
+         .withName(TABLE_NAME_PERSON_LOCAL_FS_CSV)
+         .withLabel("Person")
+         .withBackendName(defineLocalFilesystemBackend().getName())
+         .withPrimaryKeyField("id")
+         .withFields(defineCommonPersonTableFields())
+         .withBackendDetails(new FilesystemTableBackendDetails()
+            .withBasePath("persons-csv")
+            .withRecordFormat(RecordFormat.CSV)
+            .withCardinality(Cardinality.MANY)
+            .withGlob("*.csv")
          );
    }
 
@@ -202,13 +253,7 @@ public class TestUtils
          .withLabel("Person S3 Table")
          .withBackendName(defineS3Backend().getName())
          .withPrimaryKeyField("id")
-         .withField(new QFieldMetaData("id", QFieldType.INTEGER))
+         .withFields(defineCommonPersonTableFields())
-         .withField(new QFieldMetaData("createDate", QFieldType.DATE_TIME).withBackendName("create_date"))
-         .withField(new QFieldMetaData("modifyDate", QFieldType.DATE_TIME).withBackendName("modify_date"))
-         .withField(new QFieldMetaData("firstName", QFieldType.STRING).withBackendName("first_name"))
-         .withField(new QFieldMetaData("lastName", QFieldType.STRING).withBackendName("last_name"))
-         .withField(new QFieldMetaData("birthDate", QFieldType.DATE).withBackendName("birth_date"))
-         .withField(new QFieldMetaData("email", QFieldType.STRING))
          .withBackendDetails(new S3TableBackendDetails()
             .withRecordFormat(RecordFormat.CSV)
             .withCardinality(Cardinality.MANY)
@@ -220,7 +265,52 @@ public class TestUtils
    /*******************************************************************************
     **
     *******************************************************************************/
-   public static QSession getMockSession() throws QInstanceValidationException
+   public static QBackendMetaData defineMockBackend()
+   {
+      return (new QBackendMetaData()
+         .withBackendType("mock")
+         .withName(BACKEND_NAME_MOCK));
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   public static QTableMetaData defineMockPersonTable()
+   {
+      return (new QTableMetaData()
+         .withName(TABLE_NAME_PERSON_MOCK)
+         .withLabel("Person Mock Table")
+         .withBackendName(BACKEND_NAME_MOCK)
+         .withPrimaryKeyField("id")
+         .withFields(defineCommonPersonTableFields()));
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static QProcessMetaData defineStreamedLocalCsvToMockETLProcess() throws QException
+   {
+      QProcessMetaData qProcessMetaData = new StreamedETLProcess().defineProcessMetaData();
+      qProcessMetaData.setName(PROCESS_NAME_STREAMED_ETL);
+      QBackendStepMetaData backendStep  = qProcessMetaData.getBackendStep(StreamedETLProcess.FUNCTION_NAME_ETL);
+      backendStep.setCode(new QCodeReference(StreamedETLFilesystemBackendStep.class));
+
+      backendStep.getInputMetaData().getFieldThrowing(StreamedETLProcess.FIELD_SOURCE_TABLE).setDefaultValue(TABLE_NAME_PERSON_LOCAL_FS_CSV);
+      backendStep.getInputMetaData().getFieldThrowing(StreamedETLProcess.FIELD_DESTINATION_TABLE).setDefaultValue(TABLE_NAME_PERSON_MOCK);
+
+      return (qProcessMetaData);
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   public static QSession getMockSession() throws QException
    {
       MockAuthenticationModule mockAuthenticationModule = new MockAuthenticationModule();
       return (mockAuthenticationModule.createSession(defineInstance(), null));

qqq-backend-module-filesystem/src/test/java/com/kingsrook/qqq/backend/module/filesystem/local/FilesystemBackendModuleTest.java

@@ -70,7 +70,7 @@ public class FilesystemBackendModuleTest
    public void testDeleteFile() throws Exception
    {
       QInstance      qInstance = TestUtils.defineInstance();
-      QTableMetaData table     = qInstance.getTable(TestUtils.TABLE_NAME_PERSON_LOCAL_FS);
+      QTableMetaData table     = qInstance.getTable(TestUtils.TABLE_NAME_PERSON_LOCAL_FS_JSON);
 
       /////////////////////////////////////////////////////////////////////////////////////////////
       // first list the files - then delete one, then re-list, and assert that we have one fewer //
@@ -94,7 +94,7 @@ public class FilesystemBackendModuleTest
    public void testDeleteFileDoesNotExist() throws Exception
    {
       QInstance      qInstance = TestUtils.defineInstance();
-      QTableMetaData table     = qInstance.getTable(TestUtils.TABLE_NAME_PERSON_LOCAL_FS);
+      QTableMetaData table     = qInstance.getTable(TestUtils.TABLE_NAME_PERSON_LOCAL_FS_JSON);
 
       /////////////////////////////////////////////////////////////////////////////////////////////////////////////////
       // first list the files - then try to delete a fake path, then re-list, and assert that we have the same count //
@@ -120,7 +120,7 @@ public class FilesystemBackendModuleTest
    public void testMoveFile() throws Exception
    {
       QInstance      qInstance = TestUtils.defineInstance();
-      QTableMetaData table     = qInstance.getTable(TestUtils.TABLE_NAME_PERSON_LOCAL_FS);
+      QTableMetaData table     = qInstance.getTable(TestUtils.TABLE_NAME_PERSON_LOCAL_FS_JSON);
       String         basePath  = ((FilesystemBackendMetaData) qInstance.getBackendForTable(table.getName())).getBasePath();
       String         subPath   = basePath + File.separator + "subdir";
 
@@ -157,7 +157,7 @@ public class FilesystemBackendModuleTest
    public void testMoveFileDoesNotExit() throws Exception
    {
       QInstance      qInstance        = TestUtils.defineInstance();
-      QTableMetaData table            = qInstance.getTable(TestUtils.TABLE_NAME_PERSON_LOCAL_FS);
+      QTableMetaData table            = qInstance.getTable(TestUtils.TABLE_NAME_PERSON_LOCAL_FS_JSON);
       String         basePath         = ((FilesystemBackendMetaData) qInstance.getBackendForTable(table.getName())).getBasePath();
       String         subPath          = basePath + File.separator + "subdir";
       List<File>     filesBeforeMove  = new AbstractFilesystemAction().listFiles(table, qInstance.getBackendForTable(table.getName()));

qqq-backend-module-filesystem/src/test/java/com/kingsrook/qqq/backend/module/filesystem/local/actions/FilesystemActionTest.java

@@ -80,7 +80,8 @@ public class FilesystemActionTest
          fail("Failed to make directories at [" + baseDirectory + "] for filesystem backend module");
       }
 
-      writePersonFiles(baseDirectory);
+      writePersonJSONFiles(baseDirectory);
+      writePersonCSVFiles(baseDirectory);
    }
 
 
@@ -88,7 +89,7 @@ public class FilesystemActionTest
    /*******************************************************************************
     ** Write some data files into the directory for the filesystem module.
     *******************************************************************************/
-   private void writePersonFiles(File baseDirectory) throws IOException
+   private void writePersonJSONFiles(File baseDirectory) throws IOException
    {
       String fullPath = baseDirectory.getAbsolutePath();
       if (TestUtils.defineLocalFilesystemJSONPersonTable().getBackendDetails() instanceof FilesystemTableBackendDetails details)
@@ -118,6 +119,38 @@ public class FilesystemActionTest
 
 
 
+   /*******************************************************************************
+    ** Write some data files into the directory for the filesystem module.
+    *******************************************************************************/
+   private void writePersonCSVFiles(File baseDirectory) throws IOException
+   {
+      String fullPath = baseDirectory.getAbsolutePath();
+      if (TestUtils.defineLocalFilesystemCSVPersonTable().getBackendDetails() instanceof FilesystemTableBackendDetails details)
+      {
+         if (StringUtils.hasContent(details.getBasePath()))
+         {
+            fullPath += File.separatorChar + details.getBasePath();
+         }
+      }
+      fullPath += File.separatorChar;
+
+      String csvData1 = """
+         "id","createDate","modifyDate","firstName","lastName","birthDate","email"
+         "1","2021-10-26 14:39:37","2021-10-26 14:39:37","John","Doe","1981-01-01","john@kingsrook.com"
+         "2","2022-06-17 14:52:59","2022-06-17 14:52:59","Jane","Smith","1982-02-02","jane@kingsrook.com"
+         """;
+      FileUtils.writeStringToFile(new File(fullPath + "FILE-1.csv"), csvData1);
+
+      String csvData2 = """
+         "id","createDate","modifyDate","firstName","lastName","birthDate","email"
+         "3","2021-11-27 15:40:38","2021-11-27 15:40:38","Homer","S","1983-03-03","homer.s@kingsrook.com"
+         "4","2022-07-18 15:53:00","2022-07-18 15:53:00","Marge","S","1984-04-04","marge.s@kingsrook.com"
+         "5","2022-11-11 12:00:00","2022-11-12 13:00:00","Bart","S","1985-05-05","bart.s@kingsrook.com\""""; // intentionally no \n at EOL here
+      FileUtils.writeStringToFile(new File(fullPath + "FILE-2.csv"), csvData2);
+   }
+
+
+
    /*******************************************************************************
     **
     *******************************************************************************/

qqq-backend-module-filesystem/src/test/java/com/kingsrook/qqq/backend/module/filesystem/local/actions/FilesystemQueryActionTest.java

@@ -71,7 +71,7 @@ public class FilesystemQueryActionTest extends FilesystemActionTest
       QueryInput queryInput = new QueryInput();
       QInstance  instance   = TestUtils.defineInstance();
 
-      QTableMetaData table = instance.getTable(TestUtils.TABLE_NAME_PERSON_LOCAL_FS);
+      QTableMetaData table = instance.getTable(TestUtils.TABLE_NAME_PERSON_LOCAL_FS_JSON);
       table.withCustomizer(FilesystemBackendModuleInterface.CUSTOMIZER_FILE_POST_FILE_READ, new QCodeReference()
          .withName(ValueUpshifter.class.getName())
          .withCodeType(QCodeType.JAVA)

qqq-backend-module-filesystem/src/test/java/com/kingsrook/qqq/backend/module/filesystem/local/model/metadata/FilesystemBackendMetaDataTest.java

@@ -24,7 +24,7 @@ package com.kingsrook.qqq.backend.module.filesystem.local.model.metadata;
 
 import java.io.IOException;
 import com.kingsrook.qqq.backend.core.adapters.QInstanceAdapter;
-import com.kingsrook.qqq.backend.core.exceptions.QInstanceValidationException;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
 import com.kingsrook.qqq.backend.core.utils.JsonUtils;
 import com.kingsrook.qqq.backend.module.filesystem.TestUtils;
@@ -44,7 +44,7 @@ class FilesystemBackendMetaDataTest
     ** Test that an instance can be serialized as expected
     *******************************************************************************/
    @Test
-   public void testSerializingToJson() throws QInstanceValidationException
+   public void testSerializingToJson() throws QException
    {
       TestUtils.resetTestInstanceCounter();
       QInstance qInstance = TestUtils.defineInstance();
@@ -62,7 +62,7 @@ class FilesystemBackendMetaDataTest
     ** Test that an instance can be deserialized as expected
     *******************************************************************************/
    @Test
-   public void testDeserializingFromJson() throws IOException, QInstanceValidationException
+   public void testDeserializingFromJson() throws IOException, QException
    {
       QInstanceAdapter qInstanceAdapter = new QInstanceAdapter();
 

qqq-backend-module-filesystem/src/test/java/com/kingsrook/qqq/backend/module/filesystem/processes/implementations/etl/basic/BasicETLCleanupSourceFilesStepTest.java

@@ -198,7 +198,7 @@ public class BasicETLCleanupSourceFilesStepTest
       runBackendStepInput.setProcessName(qProcessMetaData.getName());
       // runFunctionRequest.setRecords(records);
       runBackendStepInput.setSession(TestUtils.getMockSession());
-      runBackendStepInput.addValue(BasicETLProcess.FIELD_SOURCE_TABLE, TestUtils.TABLE_NAME_PERSON_LOCAL_FS);
+      runBackendStepInput.addValue(BasicETLProcess.FIELD_SOURCE_TABLE, TestUtils.TABLE_NAME_PERSON_LOCAL_FS_JSON);
       runBackendStepInput.addValue(BasicETLProcess.FIELD_DESTINATION_TABLE, TestUtils.TABLE_NAME_PERSON_S3);
       runBackendStepInput.addValue(BasicETLCollectSourceFileNamesStep.FIELD_SOURCE_FILE_PATHS, StringUtils.join(",", filePathsSet));
 
@@ -219,7 +219,7 @@ public class BasicETLCleanupSourceFilesStepTest
    private String getRandomFilePathPersonTable(QInstance qInstance)
    {
       FilesystemBackendMetaData     backend        = (FilesystemBackendMetaData) qInstance.getBackend(TestUtils.BACKEND_NAME_LOCAL_FS);
-      FilesystemTableBackendDetails backendDetails = (FilesystemTableBackendDetails) qInstance.getTable(TestUtils.TABLE_NAME_PERSON_LOCAL_FS).getBackendDetails();
+      FilesystemTableBackendDetails backendDetails = (FilesystemTableBackendDetails) qInstance.getTable(TestUtils.TABLE_NAME_PERSON_LOCAL_FS_JSON).getBackendDetails();
       String                        tablePath      = backend.getBasePath() + File.separator + backendDetails.getBasePath();
       String                        filePath       = tablePath + File.separator + UUID.randomUUID();
       return filePath;

qqq-backend-module-filesystem/src/test/java/com/kingsrook/qqq/backend/module/filesystem/processes/implementations/etl/streamed/StreamedETLFilesystemBackendStepTest.java

@@ -0,0 +1,62 @@
+/*
+ * 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.module.filesystem.processes.implementations.etl.streamed;
+
+
+import com.kingsrook.qqq.backend.core.actions.processes.RunProcessAction;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessOutput;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.utils.ValueUtils;
+import com.kingsrook.qqq.backend.module.filesystem.TestUtils;
+import com.kingsrook.qqq.backend.module.filesystem.local.actions.FilesystemActionTest;
+import com.kingsrook.qqq.backend.module.filesystem.processes.implementations.etl.basic.BasicETLCollectSourceFileNamesStep;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/*******************************************************************************
+ ** Unit test for StreamedETLFilesystemBackendStep
+ *******************************************************************************/
+class StreamedETLFilesystemBackendStepTest extends FilesystemActionTest
+{
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   @Test
+   void runFullProcess() throws Exception
+   {
+      QInstance qInstance = TestUtils.defineInstance();
+
+      RunProcessInput runProcessInput = new RunProcessInput(qInstance);
+      runProcessInput.setSession(TestUtils.getMockSession());
+      runProcessInput.setProcessName(TestUtils.PROCESS_NAME_STREAMED_ETL);
+
+      RunProcessOutput output = new RunProcessAction().execute(runProcessInput);
+      String sourceFilePaths = ValueUtils.getValueAsString(output.getValues().get(BasicETLCollectSourceFileNamesStep.FIELD_SOURCE_FILE_PATHS));
+      assertThat(sourceFilePaths)
+         .contains("FILE-1.csv")
+         .contains("FILE-2.csv");
+   }
+
+}

qqq-backend-module-filesystem/src/test/java/com/kingsrook/qqq/backend/module/filesystem/s3/actions/S3QueryActionTest.java

@@ -23,7 +23,6 @@ package com.kingsrook.qqq.backend.module.filesystem.s3.actions;
 
 
 import com.kingsrook.qqq.backend.core.exceptions.QException;
-import com.kingsrook.qqq.backend.core.exceptions.QInstanceValidationException;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryOutput;
 import com.kingsrook.qqq.backend.module.filesystem.TestUtils;
@@ -60,7 +59,7 @@ public class S3QueryActionTest extends BaseS3Test
    /*******************************************************************************
     **
     *******************************************************************************/
-   private QueryInput initQueryRequest() throws QInstanceValidationException
+   private QueryInput initQueryRequest() throws QException
    {
       QueryInput queryInput = new QueryInput();
       queryInput.setInstance(TestUtils.defineInstance());

qqq-backend-module-filesystem/src/test/java/com/kingsrook/qqq/backend/module/filesystem/s3/model/metadata/S3BackendMetaDataTest.java

@@ -22,9 +22,8 @@
 package com.kingsrook.qqq.backend.module.filesystem.s3.model.metadata;
 
 
-import java.io.IOException;
 import com.kingsrook.qqq.backend.core.adapters.QInstanceAdapter;
-import com.kingsrook.qqq.backend.core.exceptions.QInstanceValidationException;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
 import com.kingsrook.qqq.backend.core.utils.JsonUtils;
 import com.kingsrook.qqq.backend.module.filesystem.TestUtils;
@@ -44,7 +43,7 @@ class S3BackendMetaDataTest
     ** Test that an instance can be serialized as expected
     *******************************************************************************/
    @Test
-   public void testSerializingToJson() throws QInstanceValidationException
+   public void testSerializingToJson() throws QException
    {
       TestUtils.resetTestInstanceCounter();
       QInstance qInstance = TestUtils.defineInstance();
@@ -62,7 +61,7 @@ class S3BackendMetaDataTest
     ** Test that an instance can be deserialized as expected
     *******************************************************************************/
    @Test
-   public void testDeserializingFromJson() throws IOException, QInstanceValidationException
+   public void testDeserializingFromJson() throws Exception
    {
       QInstanceAdapter qInstanceAdapter = new QInstanceAdapter();
 
@@ -71,6 +70,8 @@ class S3BackendMetaDataTest
 
       QInstance deserialized = qInstanceAdapter.jsonToQInstanceIncludingBackends(json);
       assertThat(deserialized.getBackends()).usingRecursiveComparison()
+         // TODO seeing occassional flaps on this field - where it can be null 1 out of 10 runs... unclear why.
+         // .ignoringFields("mock.backendType")
          .isEqualTo(qInstance.getBackends());
    }
 }