remove overwriting the inputs and fix the issue in flytekit

Signed-off-by: pmahindrakar-oss <[email protected]>

flytepropeller/pkg/controller/nodes/array/handler.go

@@ -191,12 +191,10 @@ func (a *arrayNodeHandler) Handle(ctx context.Context, nCtx interfaces.NodeExecu
 		}
 
 		size := -1
-		containsOffloadedLiteral := false
 		for _, variable := range literalMap.Literals {
 			if variable.GetOffloadedMetadata() != nil {
 				// variable will be overwritten with the contents of the offloaded data which contains the actual large literal.
 				// We need this for the map task to be able to create the subNodeSpec
-				containsOffloadedLiteral = true
 				err := common.ReadLargeLiteral(ctx, nCtx.DataStore(), variable)
 				if err != nil {
 					return handler.DoTransition(handler.TransitionTypeEphemeral,
@@ -219,16 +217,6 @@ func (a *arrayNodeHandler) Handle(ctx context.Context, nCtx interfaces.NodeExecu
 			}
 		}
 
-		// At this point if we find input to map task was an offloaded literal, then we overwrite the input file with the actual data
-		// This is due to the fact the flytekit downloads the entire data and each node indexes into the data to get the specific
-		// input that the node needs to work on.
-		if containsOffloadedLiteral {
-			inputFile := v1alpha1.GetInputsFile(nCtx.NodeStatus().GetDataDir())
-			if err := nCtx.DataStore().WriteProtobuf(ctx, inputFile, storage.Options{}, literalMap); err != nil {
-				return handler.UnknownTransition, err
-			}
-		}
-
 		if size == -1 {
 			return handler.DoTransition(handler.TransitionTypeEphemeral,
 				handler.PhaseInfoFailure(idlcore.ExecutionError_USER, errors.InvalidArrayLength, "no input array provided", nil),