to-close: added long-running materialization support to wren #374

epinzur · 2023-05-23T08:57:35Z

No description provided.

epinzur · 2023-05-23T14:55:44Z

wren/compute/prepare.go

-			preparedFile.Path = preparedFile.Path
-			subLogger.Debug().Interface("prepared_file", preparedFile).Msg("these paths should be URIs")
-		}
-


this code does nothing, removing it

epinzur · 2023-05-23T14:58:45Z

wren/compute/interface.go

+	ReconcileMaterialzations(ctx context.Context) error
+	RunMaterializations(ctx context.Context, owner *ent.Owner)
+	StartMaterialization(ctx context.Context, owner *ent.Owner, materializationID string, compileResp *v1alpha.CompileResponse, destination *v1alpha.Destination) error
+	StopMaterialization(ctx context.Context, materializationID string) error


re-org and added a few methods in the materialization section

epinzur · 2023-05-23T15:00:14Z

wren/compute/manager.go

@@ -511,7 +515,7 @@ func (m *Manager) processMaterializations(requestCtx context.Context, owner *ent
 		return nil
 	}

-	materializations, err := m.materializationClient.GetAllMaterializations(ctx, owner)
+	materializations, err := m.materializationClient.GetMaterializationsBySourceType(ctx, owner, materialization.SourceTypeFiles)


this kicks off the old-style materializations managed by wren running compute at data-load time. Now only do this on materializations that are based on file sources.

I wonder if we could rip this code out at some point soon. Maybe we make a note somewhere to do so

epinzur · 2023-05-23T15:01:25Z

wren/compute/materialization.go

+	"github.com/kaskada-ai/kaskada/wren/ent"
+	"github.com/kaskada-ai/kaskada/wren/ent/materialization"
+	"github.com/rs/zerolog/log"
+)


wrappers around calling the new methods in the compute engine for managing long-lived materializations

epinzur · 2023-05-23T15:02:09Z

wren/compute/materialization.go

+	return statusResponse.Progress, nil
+}
+
+func (m *Manager) ReconcileMaterialzations(ctx context.Context) error {


running this should make sure that all materializations in the DB are running in the compute engine

Can you add this as a comment - or can we choose a more descriptive name?
I see it's being run periodically, can you clarify why?

epinzur · 2023-05-23T15:02:39Z

wren/compute/query_context.go

@@ -104,6 +104,7 @@ func convertKaskadaTableToComputeTable(kaskadaTable *ent.KaskadaTable) *v1alpha.
 			TimeColumnName:  kaskadaTable.TimeColumnName,
 			GroupColumnName: kaskadaTable.EntityKeyColumnName,
 			Grouping:        kaskadaTable.GroupingID,
+			Source:          kaskadaTable.Source,


pass table source along to compute

epinzur · 2023-05-23T15:02:55Z

wren/ent/schema/materialization.go

@@ -31,6 +31,7 @@ func (Materialization) Fields() []ent.Field {
 		field.Bytes("slice_request").GoType(&v1alpha.SliceRequest{}).Immutable(),
 		field.Bytes("analysis").GoType(&v1alpha.Analysis{}).Immutable(),
 		field.Int64("data_version_id"),
+		field.Enum("source_type").Values("unspecified", "files", "streams").Default("unspecified"),


add a materialization source_type to the db

epinzur · 2023-05-23T15:03:39Z

wren/internal/materialization_client.go

+	return materializations, nil
+}
+
+func (c *materializationClient) GetAllMaterializationsBySourceType(ctx context.Context, sourceType materialization.SourceType) ([]*ent.Materialization, error) {


same as above, but for all owners/clients instead of just one

epinzur · 2023-05-23T15:04:24Z

wren/service/materialization.go

+			return nil, customerrors.NewInvalidArgumentErrorWithCustomText("cannot materialize tables from different source types")
+		}
+	}
+


protection around creating materializations with mixed source types

epinzur · 2023-05-23T15:04:53Z

wren/service/materialization.go

+	switch sourceType {
+	case materialization.SourceTypeFiles:
+		subLogger.Debug().Msg("running materializations")
+		s.computeManager.RunMaterializations(ctx, owner)


if is file-backed materialization, run as before

epinzur · 2023-05-23T15:05:16Z

wren/service/materialization.go

+		s.computeManager.RunMaterializations(ctx, owner)
+	case materialization.SourceTypeStreams:
+		subLogger.Debug().Msg("adding materialization to compute")
+		err := s.computeManager.StartMaterialization(ctx, owner, createdMaterialization.ID.String(), compileResp, createdMaterialization.Destination)


if is stream backed materialization, attempt to start it on compute

epinzur · 2023-05-23T15:06:17Z

wren/service/materialization_test.go

+				})
+				Expect(err).Should(HaveOccurred())
+				Expect(err.Error()).Should(Equal("cannot materialize tables from different source types"))
+				Expect(response).Should(BeNil())


test of protection around materializing from multiple source types

jordanrfrazier · 2023-05-23T20:10:39Z

wren/compute/manager.go

+		store:                   *objectStoreClient,
+		tableStore:              tableStore,
+		tr:                      otel.Tracer("ComputeManager"),
+		runningMaterailizations: map[string]struct{}{},


jordanrfrazier · 2023-05-23T20:10:56Z

wren/compute/manager.go

@@ -52,22 +53,25 @@ type Manager struct {
 	store                 client.ObjectStoreClient
 	tableStore            store.TableStore
 	tr                    trace.Tracer
+
+	runningMaterailizations map[string]struct{}


wren/compute/interface.go

jordanrfrazier · 2023-05-23T20:12:15Z

wren/compute/manager.go

@@ -511,7 +515,7 @@ func (m *Manager) processMaterializations(requestCtx context.Context, owner *ent
 		return nil
 	}

-	materializations, err := m.materializationClient.GetAllMaterializations(ctx, owner)
+	materializations, err := m.materializationClient.GetMaterializationsBySourceType(ctx, owner, materialization.SourceTypeFiles)


I wonder if we could rip this code out at some point soon. Maybe we make a note somewhere to do so

jordanrfrazier · 2023-05-23T20:13:17Z

wren/compute/materialization.go

+func (m *Manager) StartMaterialization(ctx context.Context, owner *ent.Owner, materializationID string, compileResp *v1alpha.CompileResponse, destination *v1alpha.Destination) error {
+	subLogger := log.Ctx(ctx).With().Str("method", "manager.StartMaterialization").Str("materialization_id", materializationID).Logger()
+
+	tables, err := m.getMaterializationTables(ctx, owner, compileResp)


Should we start moving away from Tables and preferring Input or Source?

jordanrfrazier · 2023-05-23T20:52:46Z

wren/compute/materialization.go

+	return statusResponse.Progress, nil
+}
+
+func (m *Manager) ReconcileMaterialzations(ctx context.Context) error {


Can you add this as a comment - or can we choose a more descriptive name?
I see it's being run periodically, can you clarify why?

jordanrfrazier · 2023-05-23T20:53:38Z

wren/compute/materialization.go

+		return err
+	}
+
+	newRunningMaterailizations := make(map[string]struct{})


jordanrfrazier · 2023-05-23T20:56:19Z

wren/compute/materialization.go

+		}
+
+		if isRunning {
+			newRunningMaterailizations[materializationID] = struct{}{}


Hm, what is the struct{}{} holding?

jordanrfrazier · 2023-05-23T21:10:10Z

wren/compute/materialization.go

+		}
+	}
+
+	// find all materializations that were running in the previous iteration but don't exist anymore


Can you elaborate on this logic? What is a previous iteration in this context

jordanrfrazier · 2023-05-23T21:27:39Z

wren/service/materialization.go

 	if err != nil {
 		return nil, err
 	}

-	err = s.materializationClient.DeleteMaterialization(ctx, owner, materialization)
+	if foundMaterialization.SourceType == materialization.SourceTypeStreams {


I assume in the future we'll also have a way to have a long-running file backed materialization, so maybe a todo here

@jordanrfrazier

@jordanrfrazier this is a new branch than the previous PR: #374 which I'm going to close.

initial update of create materializations

738c784

github-actions bot added the wren label May 23, 2023

epinzur commented May 23, 2023

View reviewed changes

implemented wren management of long-lived materializations in sparrow

f88262a

epinzur force-pushed the wren/pulsar2 branch from 8336015 to f88262a Compare May 23, 2023 14:58

epinzur commented May 23, 2023

View reviewed changes

epinzur requested a review from jordanrfrazier May 23, 2023 15:06

epinzur changed the title ~~draft: add long-running materialization support to wren~~ added long-running materialization support to wren May 23, 2023

jordanrfrazier reviewed May 23, 2023

View reviewed changes

refactored a bunch of compute stuff

725e642

epinzur changed the title ~~added long-running materialization support to wren~~ to-close: added long-running materialization support to wren May 25, 2023

epinzur mentioned this pull request May 25, 2023

added long-running materialization support to wren #387

Merged

epinzur closed this May 25, 2023

epinzur deleted the wren/pulsar2 branch May 25, 2023 16:49

epinzur added a commit that referenced this pull request May 25, 2023

added long-running materialization support to wren (#387)

a1b240d

@jordanrfrazier this is a new branch than the previous PR: #374 which I'm going to close.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

to-close: added long-running materialization support to wren #374

to-close: added long-running materialization support to wren #374

epinzur commented May 23, 2023

epinzur May 23, 2023

epinzur May 23, 2023

epinzur May 23, 2023 •

edited

Loading

jordanrfrazier May 23, 2023

epinzur May 23, 2023

epinzur May 23, 2023

jordanrfrazier May 23, 2023 •

edited

Loading

epinzur May 23, 2023

epinzur May 23, 2023

epinzur May 23, 2023

epinzur May 23, 2023

epinzur May 23, 2023

epinzur May 23, 2023

epinzur May 23, 2023

jordanrfrazier May 23, 2023

jordanrfrazier May 23, 2023

jordanrfrazier May 23, 2023

jordanrfrazier May 23, 2023

jordanrfrazier May 23, 2023 •

edited

Loading

jordanrfrazier May 23, 2023

jordanrfrazier May 23, 2023

jordanrfrazier May 23, 2023

jordanrfrazier May 23, 2023

to-close: added long-running materialization support to wren #374

to-close: added long-running materialization support to wren #374

Conversation

epinzur commented May 23, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

epinzur May 23, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jordanrfrazier May 23, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jordanrfrazier May 23, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

epinzur May 23, 2023 •

edited

Loading

jordanrfrazier May 23, 2023 •

edited

Loading

jordanrfrazier May 23, 2023 •

edited

Loading