support StreamingTransform scripts in init, scan, and deploy

.github/workflows/e2e-smoke.yml

@@ -37,6 +37,13 @@ jobs:
         with:
           python-version: '3.11'
 
+      # `script run` starts a local Spark session. PySpark supports Java 8/11/17 and
+      # fails to launch on the runner's default JDK (24+ rejects its Security Manager flag).
+      - uses: actions/setup-java@v4
+        with:
+          distribution: temurin
+          java-version: '17'
+
       - name: Install Python SDK
         shell: bash
         env:

README.md

@@ -290,6 +290,11 @@ DESCRIPTION
   Scans Python files in an initialized Data Code Extension package directory to identify required permissions and
   dependencies. Updates the config.json and requirements.txt files based on the code analysis.
 
+  For a script that calls the streaming delta methods (read_dlo_deltas or read_dmo_deltas, and write_dlo_deltas or
+  write_dmo_deltas), scan writes a streaming config instead. The read source can't be detected from the code, so scan
+  keeps streamingSource.name from config.json and uses it as the only read permission. Write permissions come from the
+  delta writes.
+
 EXAMPLES
   Scan with a custom entrypoint file:
 
@@ -481,6 +486,10 @@ FLAG DESCRIPTIONS
 
     Which transform feature this script is used in: BatchTransform (a bounded, batch data transform; the default) or
     StreamingTransform (a streaming data transform over a DLO/DMO change feed). Defaults to BatchTransform.
+
+    StreamingTransform starts from the SDK's streaming example entrypoint, adds an empty streamingSource to config.json,
+    and writes a config.yaml that switches 'run' to the local delta reader and writer. Requires
+    salesforce-data-customcode 7.0.0 or later.
 ```
 
 _See code: [src/commands/data-code-extension/script/init.ts](https://github.com/salesforcecli/plugin-data-code-extension/blob/1.6.3/src/commands/data-code-extension/script/init.ts)_
@@ -566,6 +575,11 @@ DESCRIPTION
   Scans Python files in an initialized Data Code Extension package directory to identify required permissions and
   dependencies. Updates the config.json and requirements.txt files based on the code analysis.
 
+  For a script that calls the streaming delta methods (read_dlo_deltas or read_dmo_deltas, and write_dlo_deltas or
+  write_dmo_deltas), scan writes a streaming config instead. The read source can't be detected from the code, so scan
+  keeps streamingSource.name from config.json and uses it as the only read permission. Write permissions come from the
+  delta writes.
+
 EXAMPLES
   Scan with a custom entrypoint file:
 

messages/datacodeBinaryExecutor.md

@@ -8,6 +8,23 @@ Could not locate the datacustomcode init template at '%s'.
 - Run 'pip show salesforce-data-customcode' to confirm the install location
 - Reinstall with 'pip install --upgrade salesforce-data-customcode' if the templates directory is missing
 
+# error.initRunConfigInvalid
+
+The datacustomcode run config at '%s' is not valid YAML: %s
+
+# error.initStreamingTemplateNotFound
+
+The installed salesforce-data-customcode package (version %s) doesn't include the streaming template at '%s'. StreamingTransform needs version 7.0.0 or later.
+
+# actions.initStreamingTemplateNotFound
+
+- Upgrade with 'pip install --upgrade salesforce-data-customcode'
+- Run 'pip show salesforce-data-customcode' to confirm the installed version
+
+# error.initInvalidScriptFeature
+
+Invalid use-in-feature '%s' for a script package. Valid values: %s.
+
 # error.scanEntrypointMissing
 
 Entrypoint file not found at '%s'.

messages/deploy.md

@@ -156,9 +156,25 @@ The --%s flag value exceeds the maximum length of %s characters (%s provided).
 
 API name '%s' was sanitized to '%s'.
 
-# info.inferredFeature
+# info.deploymentType
 
-Inferred feature: %s
+Deployment type: %s (%s)
+
+# info.deploymentTypeReason.batch
+
+config.json has no streamingSource
+
+# info.deploymentTypeReason.streaming
+
+config.json streamingSource: %s %s
+
+# info.deploymentTypeReason.function
+
+inferred from the entrypoint signature
+
+# info.cpuSize
+
+CPU size: %s
 
 # info.creatingDeployment
 
@@ -174,7 +190,15 @@ Uploading deployment package...
 
 # info.waitingForDeployment
 
-Waiting for deployment to complete...
+Waiting for deployment to complete. Package extraction can take several minutes.
+
+# info.waitingForDeploymentScriptNote
+
+Stopping this command doesn't cancel the deployment, but the data transform is only created after the deployment finishes. If you stop early, the data transform is not created.
+
+# info.deploymentStillWaiting
+
+Still waiting (status: %s, %s elapsed)...
 
 # info.deploymentStatusPolled
 
@@ -221,6 +245,31 @@ config.json at %s is missing required fields: %s.
 
 config.json at %s is invalid: %s.
 
+# warning.streamingEntrypointBatchConfig
+
+%s calls the streaming delta APIs (read_dlo_deltas, write_dlo_deltas, ...), but config.json has no streamingSource, so this deploys as a BatchTransform. If this is meant to be a streaming transform, run 'sf data-code-extension script scan' and set streamingSource.name, then deploy again.
+
+# error.streamingSourceNameEmpty
+
+streamingSource.name is empty in config.json at %s. A streaming transform needs the DLO or DMO it reads from.
+
+# actions.streamingSourceNameEmpty
+
+- Set streamingSource.name in config.json to the DLO or DMO the streaming transform reads from.
+- Run 'sf data-code-extension script scan' to update permissions, then deploy again.
+
+# error.streamingSourceTypeInvalid
+
+streamingSource.type in config.json at %s is '%s'. It must be 'dlo' or 'dmo'.
+
+# error.streamingSourceNotInReadPermissions
+
+streamingSource.name '%s' in config.json at %s isn't listed in %s. A streaming transform can only read from a source declared in its read permissions.
+
+# actions.streamingSourceNotInReadPermissions
+
+- Run 'sf data-code-extension script scan' to sync streamingSource.type and permissions.read with streamingSource.name, then deploy again.
+
 # error.dmoRequiresDataObjects
 
 DMO transforms require 'dataObjects' in config.json describing the schema of each output DMO.

messages/init.md

@@ -83,3 +83,9 @@ Invoke option for the script package (script packages only).
 # flags.useInFeatureScript.description
 
 Which transform feature this script is used in: BatchTransform (a bounded, batch data transform; the default) or StreamingTransform (a streaming data transform over a DLO/DMO change feed). Defaults to BatchTransform.
+
+StreamingTransform starts from the SDK's streaming example entrypoint, adds an empty streamingSource to config.json, and writes a config.yaml that switches 'run' to the local delta reader and writer. Requires salesforce-data-customcode 7.0.0 or later.
+
+# warning.streamingSourceNameEmpty
+
+Set streamingSource.name in %s to the DLO or DMO the streaming transform reads from, then run 'data-code-extension script scan'.

messages/scan.md

@@ -6,6 +6,8 @@ Scan the Data Code Extension %s package for permissions and dependencies.
 
 Scans Python files in an initialized Data Code Extension package directory to identify required permissions and dependencies. Updates the config.json and requirements.txt files based on the code analysis.
 
+For a script that calls the streaming delta methods (read_dlo_deltas or read_dmo_deltas, and write_dlo_deltas or write_dmo_deltas), scan writes a streaming config instead. The read source can't be detected from the code, so scan keeps streamingSource.name from config.json and uses it as the only read permission. Write permissions come from the delta writes.
+
 # examples.script
 
 - Scan with a custom entrypoint file:
@@ -115,6 +117,14 @@ Skips the confirmation scan asks for when the rebuilt permissions no longer cove
 dataLineage references objects that are no longer in permissions (%s entries affected). The package will fail to deploy (1041/1042) until you rerun 'data-code-extension script lineage generate' or edit dataLineage. Scan doesn't change dataLineage.
 %s
 
+# warning.streamingSourceNameEmpty
+
+streamingSource.name is empty in config.json. A streaming transform must declare its read source; set streamingSource.name to the DLO or DMO the transform reads from, then scan again.
+
+# warning.ignoredBatchObjects
+
+The entrypoint calls the streaming delta APIs, so it was scanned as a streaming transform and these batch read/write objects were left out of permissions: %s. Ignore this if the batch calls only run outside DELTA_SYNC mode; otherwise use the delta APIs for them.
+
 # prompt.staleLineage
 
 Write the updated permissions to config.json anyway?

package.json

@@ -7,7 +7,8 @@
     "@salesforce/core": "^8.2.7",
     "@salesforce/sf-plugins-core": "^12",
     "entities": "^4.5.0",
-    "jszip": "^3.10.1"
+    "jszip": "^3.10.1",
+    "yaml": "^2.9.1"
   },
   "devDependencies": {
     "@oclif/plugin-command-snapshot": "^5.2.3",

src/base/deployBase.ts

@@ -153,6 +153,7 @@ export abstract class DeployBase<TFlags extends BaseDeployFlags = BaseDeployFlag
         network,
         connection,
         log: this.log.bind(this),
+        warn: this.warn.bind(this),
       });
 
       this.log(cmdMessages.getMessage('info.deploymentComplete', [result.name, result.version]));

src/base/initBase.ts

@@ -13,10 +13,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+import * as path from 'node:path';
 import { SfCommand } from '@salesforce/sf-plugins-core';
 import { Messages } from '@salesforce/core';
 import { executeNativeInit, type NativeInitResult } from '../utils/nativeInit.js';
 import { checkEnvironment } from '../utils/environmentChecker.js';
+import { PythonChecker } from '../utils/pythonChecker.js';
+import { SCRIPT_USE_IN_FEATURE_STREAMING } from '../utils/constants.js';
 import { type SharedResultProps } from './types.js';
 
 export type BaseInitFlags = {
@@ -41,11 +44,15 @@ export abstract class InitBase<TFlags extends BaseInitFlags = BaseInitFlags> ext
     try {
       const { pythonInfo, packageInfo } = await checkEnvironment(this.spinner, this.log.bind(this), messages);
 
+      // `pip show` reports site-packages even for editable installs, whose code lives elsewhere,
+      // so ask Python where the SDK really is and fall back to pip's location.
+      const sdkDir = await PythonChecker.findPackageDir(pythonInfo.command, 'datacustomcode');
+
       this.spinner.start(messages.getMessage('info.executingInit'));
       const executionResult = await executeNativeInit({
         codeType,
         packageDir,
-        pythonPackageLocation: packageInfo.location,
+        pythonPackageLocation: sdkDir ? path.dirname(sdkDir) : packageInfo.location,
         pythonPackageVersion: packageInfo.version,
         useInFeature: additionalFlags.useInFeature as string | undefined,
       });
@@ -57,6 +64,11 @@ export abstract class InitBase<TFlags extends BaseInitFlags = BaseInitFlags> ext
         this.log(messages.getMessage('info.fileCreated', [file]));
       });
 
+      // The streaming read source can't be scanned from the code, so the new config.json leaves it empty.
+      if (codeType === 'script' && additionalFlags.useInFeature === SCRIPT_USE_IN_FEATURE_STREAMING) {
+        this.warn(messages.getMessage('warning.streamingSourceNameEmpty', [executionResult.configPath]));
+      }
+
       return {
         success: true,
         pythonVersion: pythonInfo,

src/base/scanBase.ts

@@ -16,7 +16,12 @@
 import path from 'node:path';
 import { SfCommand } from '@salesforce/sf-plugins-core';
 import { Messages, SfError } from '@salesforce/core';
-import { executeNativeScan, type NativeScanResult } from '../utils/nativeScan.js';
+import {
+  executeNativeScan,
+  findIgnoredBatchObjects,
+  isStreamingSourceNameMissing,
+  type NativeScanResult,
+} from '../utils/nativeScan.js';
 import { checkEnvironment, type EnvironmentCheckResult } from '../utils/environmentChecker.js';
 import {
   countEntriesReferencing,
@@ -71,6 +76,15 @@ export abstract class ScanBase extends SfCommand<ScanResult> {
       });
       this.spinner.stop();
 
+      if (codeType === 'script' && isStreamingSourceNameMissing(executionResult.config)) {
+        this.warn(messages.getMessage('warning.streamingSourceNameEmpty'));
+      }
+
+      if (codeType === 'script') {
+        const ignored = await findIgnoredBatchObjects(executionResult.entrypoint);
+        if (ignored.length > 0) this.warn(messages.getMessage('warning.ignoredBatchObjects', [ignored.join(', ')]));
+      }
+
       if (codeType === 'script' && executionResult.dryRun) {
         const stale = findStaleLineage(executionResult.config);
         if (hasStaleLineage(stale)) this.warnStaleLineage(messages, executionResult.config, stale);