Partial fetch

cqproto/grpc.go

@@ -63,7 +63,8 @@ func (g GRPCClient) ConfigureProvider(ctx context.Context, request *ConfigurePro
 
 func (g GRPCClient) FetchResources(ctx context.Context, request *FetchResourcesRequest) (FetchResourcesStream, error) {
 	res, err := g.client.FetchResources(ctx, &internal.FetchResources_Request{
-		Resources: request.Resources,
+		Resources:              request.Resources,
+		PartialFetchingEnabled: request.PartialFetchingEnabled,
 	})
 	if err != nil {
 		return nil, err
@@ -81,9 +82,10 @@ func (g GRPCFetchResponseStream) Recv() (*FetchResourcesResponse, error) {
 		return nil, err
 	}
 	return &FetchResourcesResponse{
-		FinishedResources: resp.GetFinishedResources(),
-		ResourceCount:     resp.GetResourceCount(),
-		Error:             resp.GetError(),
+		FinishedResources:           resp.GetFinishedResources(),
+		ResourceCount:               resp.GetResourceCount(),
+		Error:                       resp.GetError(),
+		PartialFetchFailedResources: partialFetchFailedResourcesFromProto(resp.GetPartialFetchFailedResources()),
 	}, nil
 }
 
@@ -143,7 +145,7 @@ func (g *GRPCServer) ConfigureProvider(ctx context.Context, request *internal.Co
 func (g *GRPCServer) FetchResources(request *internal.FetchResources_Request, server internal.Provider_FetchResourcesServer) error {
 	return g.Impl.FetchResources(
 		server.Context(),
-		&FetchResourcesRequest{Resources: request.GetResources()},
+		&FetchResourcesRequest{Resources: request.GetResources(), PartialFetchingEnabled: request.PartialFetchingEnabled},
 		&GRPCFetchResourcesServer{server: server},
 	)
 }
@@ -154,9 +156,10 @@ type GRPCFetchResourcesServer struct {
 
 func (g GRPCFetchResourcesServer) Send(response *FetchResourcesResponse) error {
 	return g.server.Send(&internal.FetchResources_Response{
-		FinishedResources: response.FinishedResources,
-		ResourceCount:     response.ResourceCount,
-		Error:             response.Error,
+		FinishedResources:           response.FinishedResources,
+		ResourceCount:               response.ResourceCount,
+		Error:                       response.Error,
+		PartialFetchFailedResources: partialFetchFailedResourcesToProto(response.PartialFetchFailedResources),
 	})
 }
 
@@ -233,3 +236,52 @@ func tableToProto(in *schema.Table) *internal.Table {
 		},
 	}
 }
+
+func partialFetchFailedResourcesFromProto(in []*internal.PartialFetchFailedResource) []*PartialFetchFailedResource {
+	if len(in) == 0 {
+		return nil
+	}
+	failedResources := make([]*PartialFetchFailedResource, len(in))
+	for i, p := range in {
+		failedResources[i] = &PartialFetchFailedResource{
+			TableName:            p.TableName,
+			RootTableName:        p.RootTableName,
+			RootPrimaryKeyValues: p.RootPrimaryKeyValues,
+			Error:                p.Error,
+		}
+	}
+	return failedResources
+}
+
+func partialFetchFailedResourcesToProto(in []*PartialFetchFailedResource) []*internal.PartialFetchFailedResource {
+	if len(in) == 0 {
+		return nil
+	}
+	failedResources := make([]*internal.PartialFetchFailedResource, len(in))
+	for i, p := range in {
+		failedResources[i] = &internal.PartialFetchFailedResource{
+			TableName:            p.TableName,
+			RootTableName:        p.RootTableName,
+			RootPrimaryKeyValues: p.RootPrimaryKeyValues,
+			Error:                p.Error,
+		}
+	}
+	return failedResources
+}
+
+// PartialFetchToCQProto converts schema partial fetch failed resources to cq-proto partial fetch resources
+func PartialFetchToCQProto(in []schema.PartialFetchFailedResource) []*PartialFetchFailedResource {
+	if len(in) == 0 {
+		return nil
+	}
+	failedResources := make([]*PartialFetchFailedResource, len(in))
+	for i, p := range in {
+		failedResources[i] = &PartialFetchFailedResource{
+			TableName:            p.TableName,
+			RootTableName:        p.RootTableName,
+			RootPrimaryKeyValues: p.RootPrimaryKeyValues,
+			Error:                p.Error,
+		}
+	}
+	return failedResources
+}