initialCheckPointString fix

bubriks · bubriks · commit ddf72c555116 · 2024-10-15T10:16:16.000+03:00
diff --git a/python/hsfs/engine/python.py b/python/hsfs/engine/python.py
@@ -1431,7 +1431,7 @@ def _write_dataframe_kafka(
             # provide the initial_check_point as it will reduce the read amplification of materialization job
             feature_group.materialization_job.run(
                 args=feature_group.materialization_job.config.get("defaultArgs", "")
-                + initial_check_point,
+                + (f" -initialCheckPointString {initial_check_point}" if initial_check_point else ""),
                 await_termination=offline_write_options.get("wait_for_job", False),
             )
         return feature_group.materialization_job

Original file line number	Diff line number	Diff line change
`@@ -1431,7 +1431,7 @@ def _write_dataframe_kafka(`
`1431`	`1431`	`# provide the initial_check_point as it will reduce the read amplification of materialization job`
`1432`	`1432`	`feature_group.materialization_job.run(`
`1433`	`1433`	`args=feature_group.materialization_job.config.get("defaultArgs", "")`
`1434`		`- + initial_check_point,`
	`1434`	`+ + (f" -initialCheckPointString {initial_check_point}" if initial_check_point else ""),`
`1435`	`1435`	`await_termination=offline_write_options.get("wait_for_job", False),`
`1436`	`1436`	`)`
`1437`	`1437`	`return feature_group.materialization_job`