support execute timeout (#608)

* support execute timeout

* update workflow

.github/workflows/release.yaml

@@ -25,9 +25,9 @@ jobs:
  git clone https://github.com/vesoft-inc/nebula-docker-compose.git
  pushd nebula-docker-compose/
  cp ../../client/src/test/resources/docker-compose.yaml .
- docker-compose up -d
+ docker compose up -d
  sleep 30
- docker-compose ps
+ docker compose ps
  popd
  popd
 
@@ -38,9 +38,9 @@ jobs:
  pushd ca
  cp -r ../../client/src/test/resources/ssl .
  cp ../../client/src/test/resources/docker-compose-casigned.yaml .
- docker-compose -f docker-compose-casigned.yaml up -d
+ docker compose -f docker-compose-casigned.yaml up -d
  sleep 30
- docker-compose -f docker-compose-casigned.yaml ps
+ docker compose -f docker-compose-casigned.yaml ps
  popd
  popd
 
@@ -51,9 +51,9 @@ jobs:
  pushd self
  cp -r ../../client/src/test/resources/ssl .
  cp ../../client/src/test/resources/docker-compose-selfsigned.yaml .
- docker-compose -f docker-compose-selfsigned.yaml up -d
+ docker compose -f docker-compose-selfsigned.yaml up -d
  sleep 30
- docker-compose -f docker-compose-selfsigned.yaml ps
+ docker compose -f docker-compose-selfsigned.yaml ps
  popd
  popd
 

.github/workflows/snapshot.yaml

@@ -26,9 +26,9 @@ jobs:
  git clone https://github.com/vesoft-inc/nebula-docker-compose.git
  pushd nebula-docker-compose/
  cp ../../client/src/test/resources/docker-compose.yaml .
- docker-compose up -d
+ docker compose up -d
  sleep 30
- docker-compose ps
+ docker compose ps
  popd
  popd
 
@@ -39,9 +39,9 @@ jobs:
  pushd ca
  cp -r ../../client/src/test/resources/ssl .
  cp ../../client/src/test/resources/docker-compose-casigned.yaml .
- docker-compose -f docker-compose-casigned.yaml up -d
+ docker compose -f docker-compose-casigned.yaml up -d
  sleep 30
- docker-compose -f docker-compose-casigned.yaml ps
+ docker compose -f docker-compose-casigned.yaml ps
  popd
  popd
 
@@ -52,9 +52,9 @@ jobs:
  pushd self
  cp -r ../../client/src/test/resources/ssl .
  cp ../../client/src/test/resources/docker-compose-selfsigned.yaml .
- docker-compose -f docker-compose-selfsigned.yaml up -d
+ docker compose -f docker-compose-selfsigned.yaml up -d
  sleep 30
- docker-compose -f docker-compose-selfsigned.yaml ps
+ docker compose -f docker-compose-selfsigned.yaml ps
  popd
  popd
 

client/src/main/generated/com/vesoft/nebula/ErrorCode.java

@@ -50,6 +50,7 @@ public enum ErrorCode implements com.facebook.thrift.TEnum {
  E_PART_DISABLED(-34),
  E_PART_ALREADY_STARTED(-35),
  E_PART_ALREADY_STOPPED(-36),
+ E_QUERY_TIMEDOUT(-37),
  E_BAD_USERNAME_PASSWORD(-1001),
  E_SESSION_INVALID(-1002),
  E_SESSION_TIMEOUT(-1003),

client/src/main/generated/com/vesoft/nebula/storage/RequestCommon.java

@@ -29,19 +29,23 @@ public class RequestCommon implements TBase, java.io.Serializable, Cloneable, Co
  private static final TField SESSION_ID_FIELD_DESC = new TField("session_id", TType.I64, (short)1);
  private static final TField PLAN_ID_FIELD_DESC = new TField("plan_id", TType.I64, (short)2);
  private static final TField PROFILE_DETAIL_FIELD_DESC = new TField("profile_detail", TType.BOOL, (short)3);
+ private static final TField TIMEOUT_FIELD_DESC = new TField("timeout", TType.I64, (short)4);
 
  public long session_id;
  public long plan_id;
  public boolean profile_detail;
+ public long timeout;
  public static final int SESSION_ID = 1;
  public static final int PLAN_ID = 2;
  public static final int PROFILE_DETAIL = 3;
+ public static final int TIMEOUT = 4;
 
  // isset id assignments
  private static final int __SESSION_ID_ISSET_ID = 0;
  private static final int __PLAN_ID_ISSET_ID = 1;
  private static final int __PROFILE_DETAIL_ISSET_ID = 2;
- private BitSet __isset_bit_vector = new BitSet(3);
+ private static final int __TIMEOUT_ISSET_ID = 3;
+ private BitSet __isset_bit_vector = new BitSet(4);
 
  public static final Map<Integer, FieldMetaData> metaDataMap;
 
@@ -53,6 +57,8 @@ public class RequestCommon implements TBase, java.io.Serializable, Cloneable, Co
  new FieldValueMetaData(TType.I64)));
  tmpMetaDataMap.put(PROFILE_DETAIL, new FieldMetaData("profile_detail", TFieldRequirementType.OPTIONAL, 
  new FieldValueMetaData(TType.BOOL)));
+ tmpMetaDataMap.put(TIMEOUT, new FieldMetaData("timeout", TFieldRequirementType.OPTIONAL, 
+ new FieldValueMetaData(TType.I64)));
  metaDataMap = Collections.unmodifiableMap(tmpMetaDataMap);
  }
 
@@ -66,22 +72,26 @@ public RequestCommon() {
  public RequestCommon(
  long session_id,
  long plan_id,
- boolean profile_detail) {
+ boolean profile_detail,
+ long timeout) {
  this();
  this.session_id = session_id;
  setSession_idIsSet(true);
  this.plan_id = plan_id;
  setPlan_idIsSet(true);
  this.profile_detail = profile_detail;
  setProfile_detailIsSet(true);
+ this.timeout = timeout;
+ setTimeoutIsSet(true);
  }
 
  public static class Builder {
  private long session_id;
  private long plan_id;
  private boolean profile_detail;
+ private long timeout;
 
- BitSet __optional_isset = new BitSet(3);
+ BitSet __optional_isset = new BitSet(4);
 
  public Builder() {
  }
@@ -104,6 +114,12 @@ public Builder setProfile_detail(final boolean profile_detail) {
  return this;
  }
 
+ public Builder setTimeout(final long timeout) {
+ this.timeout = timeout;
+ __optional_isset.set(__TIMEOUT_ISSET_ID, true);
+ return this;
+ }
+
  public RequestCommon build() {
  RequestCommon result = new RequestCommon();
  if (__optional_isset.get(__SESSION_ID_ISSET_ID)) {
@@ -115,6 +131,9 @@ public RequestCommon build() {
  if (__optional_isset.get(__PROFILE_DETAIL_ISSET_ID)) {
  result.setProfile_detail(this.profile_detail);
  }
+ if (__optional_isset.get(__TIMEOUT_ISSET_ID)) {
+ result.setTimeout(this.timeout);
+ }
  return result;
  }
  }
@@ -132,6 +151,7 @@ public RequestCommon(RequestCommon other) {
  this.session_id = TBaseHelper.deepCopy(other.session_id);
  this.plan_id = TBaseHelper.deepCopy(other.plan_id);
  this.profile_detail = TBaseHelper.deepCopy(other.profile_detail);
+ this.timeout = TBaseHelper.deepCopy(other.timeout);
  }
 
  public RequestCommon deepCopy() {
@@ -207,6 +227,29 @@ public void setProfile_detailIsSet(boolean __value) {
  __isset_bit_vector.set(__PROFILE_DETAIL_ISSET_ID, __value);
  }
 
+ public long getTimeout() {
+ return this.timeout;
+ }
+
+ public RequestCommon setTimeout(long timeout) {
+ this.timeout = timeout;
+ setTimeoutIsSet(true);
+ return this;
+ }
+
+ public void unsetTimeout() {
+ __isset_bit_vector.clear(__TIMEOUT_ISSET_ID);
+ }
+
+ // Returns true if field timeout is set (has been assigned a value) and false otherwise
+ public boolean isSetTimeout() {
+ return __isset_bit_vector.get(__TIMEOUT_ISSET_ID);
+ }
+
+ public void setTimeoutIsSet(boolean __value) {
+ __isset_bit_vector.set(__TIMEOUT_ISSET_ID, __value);
+ }
+
  public void setFieldValue(int fieldID, Object __value) {
  switch (fieldID) {
  case SESSION_ID:
@@ -233,6 +276,14 @@ public void setFieldValue(int fieldID, Object __value) {
  }
  break;
 
+ case TIMEOUT:
+ if (__value == null) {
+ unsetTimeout();
+ } else {
+ setTimeout((Long)__value);
+ }
+ break;
+
  default:
  throw new IllegalArgumentException("Field " + fieldID + " doesn't exist!");
  }
@@ -249,6 +300,9 @@ public Object getFieldValue(int fieldID) {
  case PROFILE_DETAIL:
  return new Boolean(isProfile_detail());
 
+ case TIMEOUT:
+ return new Long(getTimeout());
+
  default:
  throw new IllegalArgumentException("Field " + fieldID + " doesn't exist!");
  }
@@ -270,12 +324,14 @@ public boolean equals(Object _that) {
 
  if (!TBaseHelper.equalsNobinary(this.isSetProfile_detail(), that.isSetProfile_detail(), this.profile_detail, that.profile_detail)) { return false; }
 
+ if (!TBaseHelper.equalsNobinary(this.isSetTimeout(), that.isSetTimeout(), this.timeout, that.timeout)) { return false; }
+
  return true;
  }
 
  @Override
  public int hashCode() {
- return Arrays.deepHashCode(new Object[] {session_id, plan_id, profile_detail});
+ return Arrays.deepHashCode(new Object[] {session_id, plan_id, profile_detail, timeout});
  }
 
  @Override
@@ -314,6 +370,14 @@ public int compareTo(RequestCommon other) {
  if (lastComparison != 0) { 
  return lastComparison;
  }
+ lastComparison = Boolean.valueOf(isSetTimeout()).compareTo(other.isSetTimeout());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ lastComparison = TBaseHelper.compareTo(timeout, other.timeout);
+ if (lastComparison != 0) { 
+ return lastComparison;
+ }
  return 0;
  }
 
@@ -352,6 +416,14 @@ public void read(TProtocol iprot) throws TException {
  TProtocolUtil.skip(iprot, __field.type);
  }
  break;
+ case TIMEOUT:
+ if (__field.type == TType.I64) {
+ this.timeout = iprot.readI64();
+ setTimeoutIsSet(true);
+ } else { 
+ TProtocolUtil.skip(iprot, __field.type);
+ }
+ break;
  default:
  TProtocolUtil.skip(iprot, __field.type);
  break;
@@ -384,6 +456,11 @@ public void write(TProtocol oprot) throws TException {
  oprot.writeBool(this.profile_detail);
  oprot.writeFieldEnd();
  }
+ if (isSetTimeout()) {
+ oprot.writeFieldBegin(TIMEOUT_FIELD_DESC);
+ oprot.writeI64(this.timeout);
+ oprot.writeFieldEnd();
+ }
  oprot.writeFieldStop();
  oprot.writeStructEnd();
  }
@@ -433,6 +510,16 @@ public String toString(int indent, boolean prettyPrint) {
  sb.append(TBaseHelper.toString(this.isProfile_detail(), indent + 1, prettyPrint));
  first = false;
  }
+ if (isSetTimeout())
+ {
+ if (!first) sb.append("," + newLine);
+ sb.append(indentStr);
+ sb.append("timeout");
+ sb.append(space);
+ sb.append(":").append(space);
+ sb.append(TBaseHelper.toString(this.getTimeout(), indent + 1, prettyPrint));
+ first = false;
+ }
  sb.append(newLine + TBaseHelper.reduceIndent(indentStr));
  sb.append(")");
  return sb.toString();

client/src/main/java/com/vesoft/nebula/client/graph/NebulaSession.java

@@ -12,6 +12,7 @@
 import com.vesoft.nebula.client.graph.net.SessionState;
 import com.vesoft.nebula.client.graph.net.SyncConnection;
 import java.io.Serializable;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -25,9 +26,9 @@ public class NebulaSession implements Serializable {
 
  private final Logger log = LoggerFactory.getLogger(this.getClass());
 
- private final long sessionID;
- private final int timezoneOffset;
- private SyncConnection connection;
+ private final long  sessionID;
+ private final int  timezoneOffset;
+ private  SyncConnection  connection;
  private final AtomicReference<SessionState> state = new AtomicReference<>();
 
  private final AtomicBoolean isReleased = new AtomicBoolean(false);
@@ -71,6 +72,25 @@ public ResultSet executeWithParameter(String stmt, Map<String, Object> parameter
  return new ResultSet(connection.executeWithParameter(sessionID, stmt, map), timezoneOffset);
  }
 
+ public ResultSet executeWithTimeout(String stmt, long timeoutMs) throws IOErrorException {
+ return executeWithParameterTimeout(stmt,
+ (Map<String, Object>) Collections.EMPTY_MAP,
+ timeoutMs);
+ }
+
+ public ResultSet executeWithParameterTimeout(String stmt,
+ Map<String, Object> parameterMap,
+ long timeoutMs) throws IOErrorException {
+ Map<byte[], Value> map = new HashMap<>();
+ parameterMap.forEach((key, value) -> map.put(key.getBytes(), Session.value2Nvalue(value)));
+ return new ResultSet(connection.executeWithParameterTimeout(sessionID,
+ stmt,
+ map,
+ timeoutMs),
+ timezoneOffset);
+
+ }
+
  public String executeJsonWithParameter(String stmt, Map<String, Object> parameterMap)
  throws IOErrorException {
  Map<byte[], Value> map = new HashMap<>();