reformatted PipeUtil by Nitish1814 · Pull Request #1183 · zinggAI/zingg

Nitish (Nitish1814) · 2025-07-16T14:37:31Z

No description provided.

Sania Goyal (sania-16) · 2025-07-16T14:48:13Z

these changes are from uc branch - we have already made basic level changes? Nitish (@Nitish1814)

Nitish (Nitish1814) · 2025-07-16T17:59:26Z

these changes are from uc branch - we have already made basic level changes? Nitish (@Nitish1814)

These are completely new changes. whatever basic changes were made for UC this will override them

Sonal (sonalgoyal)

i took a high level look and added my comments.

Sonal (sonalgoyal) · 2025-07-28T09:13:54Z


-//spark session 
-//dataset
 public interface PipeUtilBase<S, D, R, C> {


rename to IPipeUtil

shouldnt this implement IPipeReader and IPipeWriter ?

public interface PipeUtilBase<S, D, R, C> extends IPipeUtilReader<D, R, C>, IPipeUtilWriter<D, R, C>

Sonal (sonalgoyal) · 2025-07-28T09:20:58Z

+import zingg.common.client.util.DFWriter;
+import zingg.common.client.util.writer.WriterStrategy;
+
+public class JdbcWriterStrategy<D, R, C> implements WriterStrategy<D, R, C> {


i dont think we are doing anything jdbc stuff here. just call this DefaultWriterStrategy

updated, so it will be DefaultWriterStrategy which will support some out of the box formats like Snowflake etc

Sonal (sonalgoyal) · 2025-07-28T09:24:48Z

+        this.dfWriter = dfWriter;
+    }
+
+    public WriterStrategy<D, R, C> getStrategy(Pipe<D, R, C> pipe) throws NoSuchObjectException {


this will break - for snowflake and other out of the box formats we currently support

mongo, elastic etc will break too

updated, so it will be DefaultWriterStrategy which will support some out of the box formats like Snowflake etc

Sonal (sonalgoyal) · 2025-08-11T12:05:17Z

+        if (pipe.getProps().containsKey(FilePipe.LOCATION)) {
+            return reader.load(pipe.get(FilePipe.LOCATION));
+        } else {
+            return reader.load();


this is not filereadstrategy. we need a default strategy which has reader.load which is called in the other circumstances

Changed to Default read strategy

Sonal (sonalgoyal) · 2025-08-11T12:07:09Z

+        if (Pipe.FORMAT_INMEMORY.equals(pipe.getFormat())) {
+            return new InMemoryReadStrategy<>();
+        } else {
+            return new FileReadStrategy<>();


add defaultstrategy here

added default strategy instead of file read strategy

Sonal (sonalgoyal) · 2025-08-11T12:10:39Z

+import zingg.common.client.ZFrame;
+import zingg.common.client.ZinggClientException;
+
+public interface DFReader<D, R, C> {


start name with I

apply this naming convention consistently

This has been changed

Sonal (sonalgoyal) · 2025-08-11T12:19:07Z

+    public void write(ZFrame<D, R, C> frame, Pipe<D, R, C> pipe) throws Exception {
+        DFWriter<D, R, C> writer = dfWriter
+                .format(pipe.getFormat());
+        writer.setMode(pipe.getMode() != null ? pipe.getMode() : "Append");


common code across Helper, writers etc.

Common helper is present for respective readers and writers

Sonal (sonalgoyal) · 2025-08-11T12:19:43Z

+
+    DFReader<D,R,C> setSchema(String s);
+
+    ZFrame<D,R,C> load() throws ZinggClientException;


all DFReaders do not support load(). For example the filebased ones dont. We need a base interface with format and option, common to both reader and writer. Define DFReader load which takes the pipe

Its a generic facade over DataFrameReader from spark and Snowflake

Sonal (sonalgoyal) · 2025-08-12T02:50:26Z

+
+public class Helper {
+
+    public static <D, R, C> IDFReader<D, R, C> initializeReaderForPipe(Pipe<D, R, C> pipe, IDFReader<D, R, C> dFReader) {


this can be a default method in the IDFReader instead of having it separately. Or we can have an abstract class extending IDFReader and define it there. Helpers and Utils should typically be buitl for cross cutting concerns, which this is not.

Ideally there should be a super interface with format, setSchema and option which can be inherited by both IDFReader and IDFWriter.

Sonal (sonalgoyal) · 2025-08-12T02:54:33Z

+    protected ZFrame<D, R, C> readSinglePipe(Pipe<D, R, C> pipe, boolean addSource) throws ZinggClientException {
+        try {
+            LOG.warn("Reading " + pipe);
+            IDFReader<D, R, C> reader = Helper.initializeReaderForPipe(pipe, getReader());


we should abstract away these three calls and make it easier to read using a single method call.

Sonal (sonalgoyal) · 2025-08-19T07:45:26Z

+
+public class ReadStrategyFactory<D, R, C> {
+    public ReadStrategy<D, R, C> getStrategy(Pipe<D, R, C> pipe) {
+        if (Pipe.FORMAT_INMEMORY.equals(pipe.getFormat())) {


inmem is no longer part of oss, please remove

removed in-memory pipe

* reformatted PipeUtil * break reader and writer as two separate components * changes * changed variable scope * added Helper class for initializing reader * updated WriterStrategyFactory * updated PipeUtilBase * renamed interfaces names and added Helper to for initializing writer * removed loader with string * added default reader and writer strategy * removed inMemory read strategy * changed location to path property * removed in-memory pipes * made variable protected * updated strategy reader * added back compatibility for location * removed helper from reader and writer * added pipe to writer strategy * updated snowflake format * reverted back snowflake format * added constant location and path --------- Co-authored-by: Nitish <nitish.joshi1995@gmail.com>

reformatted PipeUtil

de5af12

Nitish (Nitish1814) requested a review from Sonal (sonalgoyal) July 16, 2025 14:41

break reader and writer as two separate components

841cc63

Sonal (sonalgoyal) requested changes Jul 28, 2025

View reviewed changes

Sonal (sonalgoyal) added the pr_pending_changes label Jul 29, 2025

nitish2018 added 5 commits August 3, 2025 14:21

changes

29433fc

changed variable scope

be3043d

added Helper class for initializing reader

27f40dc

updated WriterStrategyFactory

82b580a

updated PipeUtilBase

5242337

Sonal (sonalgoyal) requested changes Aug 11, 2025

View reviewed changes

renamed interfaces names and added Helper to for initializing writer

f30609c

Sonal (sonalgoyal) requested changes Aug 12, 2025

View reviewed changes

nitish2018 added 2 commits August 13, 2025 02:10

removed loader with string

657d82d

added default reader and writer strategy

2666694

Sonal (sonalgoyal) requested changes Aug 19, 2025

View reviewed changes

nitish2018 and others added 12 commits August 19, 2025 17:23

removed inMemory read strategy

6255a7f

Merge branch 'main' into pipeutil-refactor

df44085

changed location to path property

aee56b3

removed in-memory pipes

dab4327

made variable protected

67f1b65

updated strategy reader

f676c36

added back compatibility for location

61ea072

removed helper from reader and writer

3fd0c1e

added pipe to writer strategy

fb90d33

updated snowflake format

10abd7c

reverted back snowflake format

2f5ef0c

added constant location and path

a38c7a2

Nitish (Nitish1814) added pr_pending_merge and removed pr_pending_changes labels Aug 29, 2025

Sonal (sonalgoyal) merged commit bf33525 into zinggAI:main Sep 2, 2025
1 check passed


		DFReader<D,R,C> setSchema(String s);

		ZFrame<D,R,C> load() throws ZinggClientException;


		public class Helper {

		public static <D, R, C> IDFReader<D, R, C> initializeReaderForPipe(Pipe<D, R, C> pipe, IDFReader<D, R, C> dFReader) {

Uh oh!

Conversation

Nitish (Nitish1814) commented Jul 16, 2025

Uh oh!

Sania Goyal (sania-16) commented Jul 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Nitish (Nitish1814) commented Jul 16, 2025

Uh oh!

Sonal (sonalgoyal) left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Nitish (Nitish1814) Aug 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Nitish (Nitish1814) Aug 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Sania Goyal (sania-16) commented Jul 16, 2025 •

edited

Loading

Nitish (Nitish1814) Aug 11, 2025 •

edited

Loading

Nitish (Nitish1814) Aug 11, 2025 •

edited

Loading