Let retry policy to decide what to do with potentially executed non i…

…dempotent queries

conn.go

@@ -1936,3 +1936,15 @@ func (e *QueryError) Error() string {
 func (e *QueryError) Unwrap() error {
 	return e.err
 }
+
+type PotentiallyExecutedNotIdempotentError struct {
+	err error
+}
+
+func (e *PotentiallyExecutedNotIdempotentError) Error() string {
+	return fmt.Sprintf("Potentially executed not idempotent query: %s", e.err.Error())
+}
+
+func (e *PotentiallyExecutedNotIdempotentError) Unwrap() error {
+	return e.err
+}

conn_test.go

@@ -456,6 +456,10 @@ func (t *testRetryPolicy) Attempt(qry RetryableQuery) bool {
 	return qry.Attempts() <= t.NumRetries
 }
 func (t *testRetryPolicy) GetRetryType(err error) RetryType {
+	var executedErr *PotentiallyExecutedNotIdempotentError
+	if errors.As(err, &executedErr) {
+		return Rethrow
+	}
 	return Retry
 }
 

policies.go

@@ -160,6 +160,10 @@ func (s *SimpleRetryPolicy) AttemptLWT(q RetryableQuery) bool {
 }
 
 func (s *SimpleRetryPolicy) GetRetryType(err error) RetryType {
+	var executedErr *PotentiallyExecutedNotIdempotentError
+	if errors.As(err, &executedErr) {
+		return Rethrow
+	}
 	return RetryNextHost
 }
 
@@ -168,6 +172,10 @@ func (s *SimpleRetryPolicy) GetRetryType(err error) RetryType {
 // even timeouts if other clients send statements touching the same
 // partition to the original node at the same time.
 func (s *SimpleRetryPolicy) GetRetryTypeLWT(err error) RetryType {
+	var executedErr *PotentiallyExecutedNotIdempotentError
+	if errors.As(err, &executedErr) {
+		return Rethrow
+	}
 	return Retry
 }
 
@@ -208,6 +216,10 @@ func getExponentialTime(min time.Duration, max time.Duration, attempts int) time
 }
 
 func (e *ExponentialBackoffRetryPolicy) GetRetryType(err error) RetryType {
+	var executedErr *PotentiallyExecutedNotIdempotentError
+	if errors.As(err, &executedErr) {
+		return Rethrow
+	}
 	return RetryNextHost
 }
 
@@ -216,6 +228,10 @@ func (e *ExponentialBackoffRetryPolicy) GetRetryType(err error) RetryType {
 // even timeouts if other clients send statements touching the same
 // partition to the original node at the same time.
 func (e *ExponentialBackoffRetryPolicy) GetRetryTypeLWT(err error) RetryType {
+	var executedErr *PotentiallyExecutedNotIdempotentError
+	if errors.As(err, &executedErr) {
+		return Rethrow
+	}
 	return Retry
 }
 
@@ -250,6 +266,11 @@ func (d *DowngradingConsistencyRetryPolicy) Attempt(q RetryableQuery) bool {
 }
 
 func (d *DowngradingConsistencyRetryPolicy) GetRetryType(err error) RetryType {
+	var executedErr *PotentiallyExecutedNotIdempotentError
+	if errors.As(err, &executedErr) {
+		return Rethrow
+	}
+
 	switch t := err.(type) {
 	case *RequestErrUnavailable:
 		if t.Alive > 0 {

query_executor.go

@@ -163,7 +163,7 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
 					lastErr = iter.err
 
 					if customErr, ok := iter.err.(*QueryError); ok && customErr.potentiallyExecuted && !qry.IsIdempotent() {
-						return iter
+						lastErr = &PotentiallyExecutedNotIdempotentError{err: customErr.err}
 					}
 				}
 			}