feat: clear destination in iceberg

constants/constants.go

@@ -25,6 +25,7 @@ const (
 	ConfigFolder           = "CONFIG_FOLDER"
 	StatePath              = "STATE_PATH"
 	StreamsPath            = "STREAMS_PATH"
+	DifferencePath         = "DIFFERENCE_STREAMS_PATH"
 	// DestinationDatabasePrefix is used as prefix for destination database name
 	DestinationDatabasePrefix = "DESTINATION_DATABASE_PREFIX"
 	// EffectiveParquetSize is the effective size in bytes considering 256mb targeted parquet size, compression ratio as 8

destination/iceberg/iceberg.go

@@ -613,17 +613,52 @@ func (i *Iceberg) parsePartitionRegex(pattern string) error {
 	return nil
 }
 
-func (i *Iceberg) DropStreams(_ context.Context, _ []string) error {
-	logger.Info("iceberg destination not support clear destination, skipping clear operation")
+// drop streams required for clear destination
+func (i *Iceberg) DropStreams(ctx context.Context, dropStreams []types.StreamInterface) error {
+	i.options = &destination.Options{
+		ThreadID: "iceberg_destination_drop",
+	}
+	if len(dropStreams) == 0 {
+		logger.Info("No streams selected for clearing Iceberg destination, skipping operation")
+		return nil
+	}
 
-	// logger.Infof("Clearing Iceberg destination for %d selected streams: %v", len(selectedStreams), selectedStreams)
+	// server setup for dropping tables
+	server, err := newIcebergClient(i.config, []PartitionInfo{}, i.options.ThreadID, false, false, "")
+	if err != nil {
+		return fmt.Errorf("failed to setup iceberg server for dropping streams: %s", err)
+	}
 
-	// TODO: Implement Iceberg table clearing logic
-	// 1. Connect to the Iceberg catalog
-	// 2. Use Iceberg's delete API or drop/recreate the table
-	// 3. Handle any Iceberg-specific cleanup
+	// to close client properly
+	i.server = server
+	defer func() {
+		i.Close(ctx)
+	}()
+
+	logger.Infof("Starting Clear Iceberg destination for %d selected streams", len(dropStreams))
+
+	// process each stream
+	for _, stream := range dropStreams {
+		destDB := stream.GetDestinationDatabase(&i.config.IcebergDatabase)
+		destTable := stream.GetDestinationTable()
+		dropTable := fmt.Sprintf("%s.%s", destDB, destTable)
+
+		logger.Infof("Dropping Iceberg table: %s", dropTable)
+
+		request := proto.IcebergPayload{
+			Type: proto.IcebergPayload_DROP_TABLE,
+			Metadata: &proto.IcebergPayload_Metadata{
+				DestTableName: dropTable,
+				ThreadId:      i.server.serverID,
+			},
+		}
+		_, err := i.server.sendClientRequest(ctx, &request)
+		if err != nil {
+			return fmt.Errorf("failed to drop table %s: %s", dropTable, err)
+		}
+	}
 
-	// logger.Info("Successfully cleared Iceberg destination for selected streams")
+	logger.Info("Successfully cleared Iceberg destination for selected streams")
 	return nil
 }
 

destination/iceberg/olake-iceberg-java-writer/src/main/java/io/debezium/server/iceberg/IcebergUtil.java

@@ -276,4 +276,19 @@ public static int partitionId() {
     return Integer.parseInt(dtFormater.format(Instant.now()));
   }
 
+  public static boolean dropIcebergTable(String namespace, String tableName, Catalog icebergCatalog) {
+    try{
+      TableIdentifier tableID = TableIdentifier.of(namespace, tableName);
+      // Check if table exists
+      if (!icebergCatalog.tableExists(tableID)) {
+        LOGGER.warn("Table not found: {}", tableID.toString());
+        return false;
+      }
+      return icebergCatalog.dropTable(tableID, true);
+    } catch(Exception e){
+      LOGGER.error("Failed to drop table {}.{}: {}", namespace, tableName, e.getMessage());
+      throw new RuntimeException("Failed to drop table: " + namespace + "." + tableName, e);
+    }
+  }
+
 }

destination/iceberg/olake-iceberg-java-writer/src/main/java/io/debezium/server/iceberg/rpc/OlakeRowsIngester.java

@@ -62,7 +62,7 @@ public void sendRecords(IcebergPayload request, StreamObserver<RecordIngest.Reco
                 throw new Exception("Destination table name not present in metadata");
             }
 
-            if (this.icebergTable == null) {
+            if (this.icebergTable == null && request.getType() != IcebergPayload.PayloadType.DROP_TABLE) {
                 SchemaConvertor schemaConvertor = new SchemaConvertor(identifierField, schemaMetadata);
                 this.icebergTable = loadIcebergTable(TableIdentifier.of(icebergNamespace, destTableName), 
                                         schemaConvertor.convertToIcebergSchema());
@@ -111,10 +111,24 @@ public void sendRecords(IcebergPayload request, StreamObserver<RecordIngest.Reco
                     break;
 
                 case DROP_TABLE:
-                    LOGGER.warn("{} Table {} not dropped, drop table not implemented", requestId, destTableName);
-                    sendResponse(responseObserver, "Drop table not implemented");
+                    String dropTable = metadata.getDestTableName();
+                    String[] parts = dropTable.split("\\.", 2);
+                    if (parts.length != 2) {
+                        throw new IllegalArgumentException("Invalid destination table name: " + dropTable);
+                    }
+                    String namespace = parts[0], tableName = parts[1];
+
+                    LOGGER.warn("{} Dropping table {}.{}", requestId, namespace, tableName);
+
+                    boolean dropped = IcebergUtil.dropIcebergTable(namespace, tableName, icebergCatalog);
+                    if (dropped) {
+                        sendResponse(responseObserver, "Successfully dropped table " + tableName);
+                        LOGGER.info("{} Table {} dropped", requestId, tableName);
+                    } else {
+                        sendResponse(responseObserver, "Table " + tableName + " does not exist");
+                        LOGGER.warn("{} Table {} not dropped, table does not exist", requestId, tableName);
+                    }
                     break;
-
                 default:
                     throw new IllegalArgumentException("Unknown payload type: " + request.getType());
             }

destination/interface.go

@@ -32,6 +32,6 @@ type Writer interface {
 	// Need to pass olakeTimestamp as end argument to get the correct partition path based on record ingestion time.
 	EvolveSchema(ctx context.Context, globalSchema, recordsSchema any) (any, error)
 	// DropStreams is used to clear the destination before re-writing the stream
-	DropStreams(ctx context.Context, selectedStream []string) error
+	DropStreams(ctx context.Context, dropStreams []types.StreamInterface) error
 	Close(ctx context.Context) error
 }

destination/parquet/parquet.go

@@ -456,20 +456,25 @@ func (p *Parquet) getPartitionedFilePath(values map[string]any, olakeTimestamp t
 	return filepath.Join(p.basePath, strings.TrimSuffix(result, "/"))
 }
 
-func (p *Parquet) DropStreams(ctx context.Context, selectedStreams []string) error {
+func (p *Parquet) DropStreams(ctx context.Context, selectedStreams []types.StreamInterface) error {
 	if len(selectedStreams) == 0 {
 		logger.Infof("Thread[%s]: no streams selected for clearing, skipping clear operation", p.options.ThreadID)
 		return nil
 	}
 
 	logger.Infof("Thread[%s]: clearing destination for %d selected streams: %v", p.options.ThreadID, len(selectedStreams), selectedStreams)
 
+	paths := make([]string, 0, len(selectedStreams))
+	for _, stream := range selectedStreams {
+		paths = append(paths, stream.GetDestinationDatabase(nil)+"."+stream.GetDestinationTable())
+	}
+
 	if p.s3Client == nil {
-		if err := p.clearLocalFiles(selectedStreams); err != nil {
+		if err := p.clearLocalFiles(paths); err != nil {
 			return fmt.Errorf("failed to clear local files: %s", err)
 		}
 	} else {
-		if err := p.clearS3Files(ctx, selectedStreams); err != nil {
+		if err := p.clearS3Files(ctx, paths); err != nil {
 			return fmt.Errorf("failed to clear S3 files: %s", err)
 		}
 	}
@@ -478,16 +483,15 @@ func (p *Parquet) DropStreams(ctx context.Context, selectedStreams []string) err
 	return nil
 }
 
-func (p *Parquet) clearLocalFiles(selectedStreams []string) error {
-	for _, streamID := range selectedStreams {
+func (p *Parquet) clearLocalFiles(paths []string) error {
+	for _, streamID := range paths {
 		parts := strings.SplitN(streamID, ".", 2)
 		if len(parts) != 2 {
 			logger.Warnf("Thread[%s]: invalid stream ID format: %s, skipping", p.options.ThreadID, streamID)
 			continue
 		}
-
-		namespace, streamName := parts[0], parts[1]
-		streamPath := filepath.Join(p.config.Path, namespace, streamName)
+		namespace, tableName := parts[0], parts[1]
+		streamPath := filepath.Join(p.config.Path, namespace, tableName)
 
 		logger.Infof("Thread[%s]: clearing local path: %s", p.options.ThreadID, streamPath)
 
@@ -506,7 +510,7 @@ func (p *Parquet) clearLocalFiles(selectedStreams []string) error {
 	return nil
 }
 
-func (p *Parquet) clearS3Files(ctx context.Context, selectedStreams []string) error {
+func (p *Parquet) clearS3Files(ctx context.Context, paths []string) error {
 	deleteS3PrefixStandard := func(filtPath string) error {
 		iter := s3manager.NewDeleteListIterator(p.s3Client, &s3.ListObjectsInput{
 			Bucket: aws.String(p.config.Bucket),
@@ -519,15 +523,15 @@ func (p *Parquet) clearS3Files(ctx context.Context, selectedStreams []string) er
 		return nil
 	}
 
-	for _, streamID := range selectedStreams {
+	for _, streamID := range paths {
 		parts := strings.SplitN(streamID, ".", 2)
 		if len(parts) != 2 {
 			logger.Warnf("Thread[%s]: invalid stream ID format: %s, skipping", p.options.ThreadID, streamID)
 			continue
 		}
+		namespace, tableName := parts[0], parts[1]
+		s3TablePath := filepath.Join(p.config.Prefix, namespace, tableName, "/")
 
-		namespace, streamName := parts[0], parts[1]
-		s3TablePath := filepath.Join(p.config.Prefix, namespace, streamName, "/")
 		logger.Debugf("Thread[%s]: clearing S3 prefix: s3://%s/%s", p.options.ThreadID, p.config.Bucket, s3TablePath)
 		if err := deleteS3PrefixStandard(s3TablePath); err != nil {
 			return fmt.Errorf("failed to clear S3 prefix %s: %s", s3TablePath, err)

destination/writers.go

@@ -83,7 +83,7 @@ func WithThreadID(threadID string) ThreadOptions {
 	}
 }
 
-func NewWriterPool(ctx context.Context, config *types.WriterConfig, syncStreams, dropStreams []string, batchSize int64) (*WriterPool, error) {
+func NewWriterPool(ctx context.Context, config *types.WriterConfig, syncStreams []string, batchSize int64) (*WriterPool, error) {
 	newfunc, found := RegisteredWriters[config.Type]
 	if !found {
 		return nil, fmt.Errorf("invalid destination type has been passed [%s]", config.Type)
@@ -99,12 +99,6 @@ func NewWriterPool(ctx context.Context, config *types.WriterConfig, syncStreams,
 		return nil, fmt.Errorf("failed to test destination: %s", err)
 	}
 
-	if dropStreams != nil {
-		if err := adapter.DropStreams(ctx, dropStreams); err != nil {
-			return nil, fmt.Errorf("failed to clear destination: %s", err)
-		}
-	}
-
 	pool := &WriterPool{
 		stats: &Stats{
 			TotalRecordsToSync: atomic.Int64{},
@@ -279,3 +273,22 @@ func (wt *WriterThread) Close(ctx context.Context) error {
 		return wt.writer.Close(ctx)
 	}
 }
+
+func ClearDestination(ctx context.Context, config *types.WriterConfig, dropStreams []types.StreamInterface) error {
+	newfunc, found := RegisteredWriters[config.Type]
+	if !found {
+		return fmt.Errorf("invalid destination type has been passed [%s]", config.Type)
+	}
+
+	adapter := newfunc()
+	if err := utils.Unmarshal(config.WriterConfig, adapter.GetConfigRef()); err != nil {
+		return err
+	}
+
+	if dropStreams != nil {
+		if err := adapter.DropStreams(ctx, dropStreams); err != nil {
+			return fmt.Errorf("failed to drop the streams: %s", err)
+		}
+	}
+	return nil
+}

drivers/abstract/abstract.go

@@ -113,7 +113,34 @@ func (a *AbstractDriver) Setup(ctx context.Context) error {
 	return a.driver.Setup(ctx)
 }
 
-// Read handles different sync modes for data retrieval
+func (a *AbstractDriver) ClearState(streams []types.StreamInterface) (*types.State, error) {
+	if a.state == nil {
+		return &types.State{}, nil
+	}
+
+	dropStreams := make(map[string]bool)
+	for _, stream := range streams {
+		dropStreams[stream.ID()] = true
+	}
+
+	// if global state exists (in case of relational sources)
+	if a.state.Global != nil && a.state.Global.Streams != nil {
+		for streamID := range dropStreams {
+			a.state.Global.Streams.Remove(streamID)
+		}
+	}
+
+	if len(a.state.Streams) > 0 {
+		for _, streamState := range a.state.Streams {
+			if dropStreams[fmt.Sprintf("%s.%s", streamState.Namespace, streamState.Stream)] {
+				streamState.HoldsValue.Store(false)
+				streamState.State = sync.Map{}
+			}
+		}
+	}
+	return a.state, nil
+}
+
 func (a *AbstractDriver) Read(ctx context.Context, pool *destination.WriterPool, backfillStreams, cdcStreams, incrementalStreams []types.StreamInterface) error {
 	// set max read connections
 	if a.driver.MaxConnections() > 0 {

protocol/check.go

@@ -37,7 +37,7 @@ var checkCmd = &cobra.Command{
 		err := func() error {
 			// If connector is not set, we are checking the destination
 			if destinationConfigPath != "not-set" {
-				_, err := destination.NewWriterPool(cmd.Context(), destinationConfig, nil, nil, batchSize)
+				_, err := destination.NewWriterPool(cmd.Context(), destinationConfig, nil, batchSize)
 				return err
 			}
 

protocol/clear.go

@@ -0,0 +1,77 @@
+package protocol
+
+import (
+	"fmt"
+
+	"github.com/datazip-inc/olake/destination"
+	"github.com/datazip-inc/olake/types"
+	"github.com/datazip-inc/olake/utils"
+	"github.com/datazip-inc/olake/utils/logger"
+	"github.com/spf13/cobra"
+)
+
+var clearCmd = &cobra.Command{
+	Use:   "clear-destination",
+	Short: "Olake clear command to clear destination data and state for selected streams",
+	PersistentPreRunE: func(_ *cobra.Command, _ []string) error {
+		if destinationConfigPath == "" {
+			return fmt.Errorf("--destination not passed")
+		} else if streamsPath == "" {
+			return fmt.Errorf("--streams not passed")
+		}
+
+		destinationConfig = &types.WriterConfig{}
+		if err := utils.UnmarshalFile(destinationConfigPath, destinationConfig, true); err != nil {
+			return err
+		}
+
+		catalog = &types.Catalog{}
+		if err := utils.UnmarshalFile(streamsPath, catalog, false); err != nil {
+			return err
+		}
+
+		state = &types.State{
+			Type: types.StreamType,
+		}
+		if statePath != "" {
+			if err := utils.UnmarshalFile(statePath, state, false); err != nil {
+				return err
+			}
+		}
+		return nil
+	},
+	RunE: func(cmd *cobra.Command, _ []string) error {
+		selectedStreamsMetadata, err := GetStreamsClassification(catalog, nil, state)
+		if err != nil {
+			return fmt.Errorf("failed to get selected streams for clearing: %w", err)
+		}
+		dropStreams := []types.StreamInterface{}
+		dropStreams = append(dropStreams, append(append(selectedStreamsMetadata.IncrementalStreams, selectedStreamsMetadata.FullLoadStreams...), selectedStreamsMetadata.CDCStreams...)...)
+		if len(dropStreams) == 0 {
+			logger.Infof("No streams selected for clearing")
+			return nil
+		}
+
+		connector.SetupState(state)
+		// clear state for selected streams
+		newState, err := connector.ClearState(dropStreams)
+		if err != nil {
+			return fmt.Errorf("error clearing state: %w", err)
+		}
+		logger.Infof("State for selected streams cleared successfully.")
+		// Setup new state after clear for connector
+		connector.SetupState(newState)
+
+		// drop/clear streams from destination
+		cerr := destination.ClearDestination(cmd.Context(), destinationConfig, dropStreams)
+		if cerr != nil {
+			return fmt.Errorf("failed to clear destination: %w", err)
+		}
+		logger.Infof("Successfully cleared destination data for selected streams.")
+		stateBytes, _ := newState.MarshalJSON()
+		logger.Infof("New saved state: %s", stateBytes)
+		// save new state in state file
+		newState.LogState()
+		return nil
+	},
+}