feat: add merge and replace merge without partition

max2max/internal/loader/factory.go

@@ -20,10 +20,10 @@ func GetLoader(name string, logger *slog.Logger) (Loader, error) {
 	// 	return NewReplaceLoader(logger), nil
 	// case REPLACE_ALL:
 	// 	return NewReplaceAllLoader(logger), nil
-	// case MERGE:
-	// 	return NewMergeLoader(logger), nil
-	// case MERGE_REPLACE:
-	// 	return NewMergeReplaceLoader(logger), nil
+	case MERGE:
+		return NewMergeLoader(logger)
+	case MERGE_REPLACE:
+		return NewReplaceMergeLoader(logger)
 	default:
 		err := fmt.Errorf("loader %s not found", name)
 		return nil, errors.WithStack(err)

max2max/internal/loader/merge.go

@@ -8,16 +8,16 @@ type mergeLoader struct {
 	logger *slog.Logger
 }
 
-func NewMergeLoader(logger *slog.Logger) *mergeLoader {
+func NewMergeLoader(logger *slog.Logger) (*mergeLoader, error) {
 	return &mergeLoader{
 		logger: logger,
-	}
+	}, nil
 }
 
-func (l *mergeLoader) GetQuery(tableID, query string) string {
-	return "-- TODO merge loader"
+func (l *mergeLoader) GetQuery(_, query string) string {
+	return query
 }
 
-func (l *mergeLoader) GetPartitionedQuery(tableID, query string, partitionName []string) string {
-	return "-- TODO merge loader"
+func (l *mergeLoader) GetPartitionedQuery(_, query string, _ []string) string {
+	return query
 }

max2max/internal/loader/replace_merge.go

@@ -0,0 +1,24 @@
+package loader
+
+import (
+	"fmt"
+	"log/slog"
+)
+
+type replaceMergeLoader struct {
+	logger *slog.Logger
+}
+
+func NewReplaceMergeLoader(logger *slog.Logger) (*replaceMergeLoader, error) {
+	return &replaceMergeLoader{
+		logger: logger,
+	}, nil
+}
+
+func (l *replaceMergeLoader) GetQuery(tableID, query string) string {
+	return fmt.Sprintf("INSERT OVERWRITE TABLE %s %s", tableID, query)
+}
+
+func (l *replaceMergeLoader) GetPartitionedQuery(tableID, query string, partitionName []string) string {
+	return "-- TODO merge replace loader"
+}