Add support cache for dynamic spec

datacatalog/go.mod

@@ -17,6 +17,7 @@ require (
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.47.0
 	go.opentelemetry.io/otel v1.24.0
 	google.golang.org/grpc v1.62.1
+	google.golang.org/protobuf v1.34.1
 	gorm.io/driver/postgres v1.5.3
 	gorm.io/driver/sqlite v1.5.4
 	gorm.io/gorm v1.25.4
@@ -131,7 +132,6 @@ require (
 	google.golang.org/genproto v0.0.0-20240123012728-ef4313101c80 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240123012728-ef4313101c80 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240123012728-ef4313101c80 // indirect
-	google.golang.org/protobuf v1.34.1 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/ini.v1 v1.66.4 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect

datacatalog/pkg/manager/impl/artifact_data_store.go

@@ -2,6 +2,7 @@
 
 import (
 	"context"
+	"fmt"
 
 	"google.golang.org/grpc/codes"
 
@@ -13,10 +14,14 @@
 )
 
 const artifactDataFile = "data.pb"
+const futureDataFile = "future.pb"
+const futureDataName = "future"
 
 // ArtifactDataStore stores and retrieves ArtifactData values in a data.pb
 type ArtifactDataStore interface {
 	PutData(ctx context.Context, artifact *datacatalog.Artifact, data *datacatalog.ArtifactData) (storage.DataReference, error)
+	PutFutureData(ctx context.Context, artifact *datacatalog.Artifact, djspec *core.DynamicJobSpec) (storage.DataReference, error)
+	GetFutureData(ctx context.Context, dataModel models.ArtifactData) (*core.DynamicJobSpec, error)
 	GetData(ctx context.Context, dataModel models.ArtifactData) (*core.Literal, error)
 	DeleteData(ctx context.Context, dataModel models.ArtifactData) error
 }
@@ -31,6 +36,11 @@
 	return m.store.ConstructReference(ctx, m.storagePrefix, dataset.GetProject(), dataset.GetDomain(), dataset.GetName(), dataset.GetVersion(), artifact.GetId(), data.GetName(), artifactDataFile)
 }
 
+func (m *artifactDataStore) getFutureDataLocation(ctx context.Context, artifact *datacatalog.Artifact) (storage.DataReference, error) {
+	dataset := artifact.GetDataset()
+	return m.store.ConstructReference(ctx, m.storagePrefix, dataset.GetProject(), dataset.GetDomain(), dataset.GetName(), dataset.GetVersion(), artifact.GetId(), futureDataName, futureDataFile)
+}
+
 // Store marshalled data in data.pb under the storage prefix
 func (m *artifactDataStore) PutData(ctx context.Context, artifact *datacatalog.Artifact, data *datacatalog.ArtifactData) (storage.DataReference, error) {
 	dataLocation, err := m.getDataLocation(ctx, artifact, data)
@@ -45,6 +55,20 @@
 	return dataLocation, nil
 }
 
+// Store marshalled future data in future.pb under the future storage prefix
+func (m *artifactDataStore) PutFutureData(ctx context.Context, artifact *datacatalog.Artifact, djspec *core.DynamicJobSpec) (storage.DataReference, error) {
+	dataLocation, err := m.getFutureDataLocation(ctx, artifact)
+	if err != nil {
+		return "", errors.NewDataCatalogErrorf(codes.Internal, "Unable to generate data location %s, err %v", dataLocation.String(), err)
+	}
+	err = m.store.WriteProtobuf(ctx, dataLocation, storage.Options{}, djspec)
+	if err != nil {
+		return "", errors.NewDataCatalogErrorf(codes.Internal, "Unable to store future artifact data in location %s, err %v", dataLocation.String(), err)
+	}
+
+	return dataLocation, nil
+}
+
 // Retrieve the literal value of the ArtifactData from its specified location
 func (m *artifactDataStore) GetData(ctx context.Context, dataModel models.ArtifactData) (*core.Literal, error) {
 	var value core.Literal
@@ -56,6 +80,18 @@
 	return &value, nil
 }
 
+// Retrieve the future data from the future storage prefix
+func (m *artifactDataStore) GetFutureData(ctx context.Context, dataModel models.ArtifactData) (*core.DynamicJobSpec, error) {
+	var djSpec core.DynamicJobSpec
+	err := m.store.ReadProtobuf(ctx, storage.DataReference(dataModel.Location), &djSpec)
+	if err != nil {
+		return nil, errors.NewDataCatalogErrorf(codes.Internal, "Unable to read future artifact data from location %s, err %v", dataModel.Location, err)
+	}
+	fmt.Printf("the dj spec: %+v\n", djSpec.GetNodes()[0])
+
+	return &djSpec, nil
+}
+
 // DeleteData removes the stored artifact data from the underlying blob storage
 func (m *artifactDataStore) DeleteData(ctx context.Context, dataModel models.ArtifactData) error {
 	if err := m.store.Delete(ctx, storage.DataReference(dataModel.Location)); err != nil {

datacatalog/pkg/manager/impl/artifact_manager.go

@@ -208,6 +208,7 @@
 				m.systemMetrics.doesNotExistCounter.Inc(ctx)
 			} else {
 				logger.Errorf(ctx, "Unable to retrieve Artifact by tag %v, err: %v", key, err)
+				m.systemMetrics.getFailureCounter.Inc(ctx)
 			}
 			return models.Artifact{}, err
 		}