Add GetOutputMetadata to the output reader

go/tasks/pluginmachinery/io/iface.go

@@ -43,6 +43,8 @@ type OutputReader interface {
 	Read(ctx context.Context) (*core.LiteralMap, *ExecutionError, error)
 	// DeckExists checks if the deck file has been generated.
 	DeckExists(ctx context.Context) (bool, error)
+	// GetOutputMetadata get the metadata from the output of tasks, such as deck URI.
+	GetOutputMetadata(ctx context.Context) map[string]string
 }
 
 // CheckpointPaths provides the paths / keys to input Checkpoints directory and an output checkpoints directory.

go/tasks/pluginmachinery/ioutils/in_memory_output_reader.go

@@ -47,6 +47,13 @@ func (r InMemoryOutputReader) DeckExists(_ context.Context) (bool, error) {
 	return r.DeckPath != nil, nil
 }
 
+func (r InMemoryOutputReader) GetOutputMetadata(_ context.Context) map[string]string {
+	if r.DeckPath == nil {
+		return map[string]string{}
+	}
+	return map[string]string{deckURIKey: r.DeckPath.String()}
+}
+
 func NewInMemoryOutputReader(literals *core.LiteralMap, DeckPath *storage.DataReference, err *io.ExecutionError) InMemoryOutputReader {
 	return InMemoryOutputReader{
 		literals: literals,

go/tasks/pluginmachinery/ioutils/in_memory_output_reader_test.go

@@ -25,6 +25,7 @@ func TestInMemoryOutputReader(t *testing.T) {
 	or := NewInMemoryOutputReader(&flyteIdlCore.LiteralMap{Literals: lt}, &deckPath, nil)
 
 	assert.Equal(t, &deckPath, or.DeckPath)
+	assert.Equal(t, deckPath.String(), or.GetOutputMetadata(context.Background())[deckURIKey])
 	ctx := context.TODO()
 
 	ok, err := or.IsError(ctx)

go/tasks/pluginmachinery/ioutils/paths.go

@@ -30,6 +30,8 @@ const (
 	// CheckpointPrefix specifies the common prefix that can be used as a directory where all the checkpoint information
 	// will be stored. This directory is under the raw output-prefix path
 	CheckpointPrefix = "_flytecheckpoints"
+
+	deckURIKey = "deck-uri"
 )
 
 // ConstructCheckpointPath returns a checkpoint path under the given `base` / rawOutputPrefix, following the conventions of

go/tasks/pluginmachinery/ioutils/remote_file_output_reader.go

@@ -123,6 +123,10 @@ func (r RemoteFileOutputReader) DeckExists(ctx context.Context) (bool, error) {
 	return md.Exists(), nil
 }
 
+func (r RemoteFileOutputReader) GetOutputMetadata(_ context.Context) map[string]string {
+	return map[string]string{deckURIKey: r.outPath.GetDeckPath().String()}
+}
+
 func NewRemoteFileOutputReader(_ context.Context, store storage.ComposedProtobufStore, outPaths io.OutputFilePaths, maxDatasetSize int64) RemoteFileOutputReader {
 	return RemoteFileOutputReader{
 		outPath:        outPaths,

go/tasks/pluginmachinery/ioutils/remote_file_output_reader_test.go

@@ -68,6 +68,7 @@ func TestReadOrigin(t *testing.T) {
 		exists, err := r.DeckExists(ctx)
 		assert.NoError(t, err)
 		assert.True(t, exists)
+		assert.Equal(t, "deck.html", r.GetOutputMetadata(ctx)[deckURIKey])
 	})
 
 	t.Run("system", func(t *testing.T) {