Add Decimal

Author: Ryan Murray

src/main/java/com/dremio/flight/Producer.java

@@ -45,6 +45,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.dremio.common.exceptions.UserException;
 import com.dremio.common.exceptions.UserRemoteException;
 import com.dremio.common.utils.protos.ExternalIdHelper;
 import com.dremio.common.utils.protos.QueryWritableBatch;
@@ -100,9 +101,11 @@ class Producer implements FlightProducer, AutoCloseable {
 
   @Override
   public void doAction(CallContext context, Action action, StreamListener<Result> resultStreamListener) {
+    throw Status.UNIMPLEMENTED.asRuntimeException();
   }
 
   private FlightInfo getInfo(CallContext callContext, FlightDescriptor descriptor, String sql) {
+    logger.info("GetFlightInfo called for sql {}", sql);
     return getInfoImpl(callContext, descriptor, sql);
   }
 
@@ -116,8 +119,9 @@ private FlightInfo getInfoImpl(CallContext callContext, FlightDescriptor descrip
 
       UserRequest request = new UserRequest(RpcType.CREATE_PREPARED_STATEMENT, req);
       Prepare prepare = new Prepare();
-      submitWork(callContext, request, prepare);
-      return prepare.getInfo(descriptor);
+
+      UserBitShared.ExternalId externalId = submitWork(callContext, request, prepare);
+      return prepare.getInfo(descriptor, externalId);
     } catch (Exception e) {
       e.printStackTrace();
       throw new RuntimeException(e);
@@ -126,13 +130,12 @@ private FlightInfo getInfoImpl(CallContext callContext, FlightDescriptor descrip
 
   @Override
   public FlightInfo getFlightInfo(CallContext callContext, FlightDescriptor descriptor) {
-    logger.info("called get flight info");
     return getInfo(callContext, descriptor, new String(descriptor.getCommand()));
   }
 
   @Override
   public Runnable acceptPut(CallContext callContext, FlightStream flightStream, StreamListener<org.apache.arrow.flight.PutResult> streamListener) {
-    throw Status.UNAVAILABLE.asRuntimeException();
+    throw Status.UNIMPLEMENTED.asRuntimeException();
   }
 
   private UserBitShared.ExternalId submitWork(CallContext callContext, UserRequest request, UserResponseHandler handler) {
@@ -143,6 +146,7 @@ private UserBitShared.ExternalId submitWork(CallContext callContext, UserRequest
       handler,
       request,
       TerminationListenerRegistry.NOOP);
+    logger.debug("Submitted job {} from flight for request with type {}", ExternalIdHelper.toQueryId(externalId), request.getType());
     return externalId;
   }
 
@@ -158,28 +162,34 @@ private class Prepare implements UserResponseHandler {
     public Prepare() {
     }
 
-    public FlightInfo getInfo(FlightDescriptor descriptor) {
+    public FlightInfo getInfo(FlightDescriptor descriptor, UserBitShared.ExternalId externalId) {
       try {
+        logger.debug("Waiting for prepared statement handle to return for job id {}", ExternalIdHelper.toQueryId(externalId));
         CreatePreparedStatementResp handle = future.get();
+        logger.debug("prepared statement handle for job id {} has returned", ExternalIdHelper.toQueryId(externalId));
         if (handle.getStatus() == RequestStatus.FAILED) {
-          throw Status.UNKNOWN.withDescription(handle.getError().getMessage()).withCause(UserRemoteException.create(handle.getError())).asRuntimeException();
+          logger.warn("prepared statement handle for job id " + ExternalIdHelper.toQueryId(externalId) + " has failed", UserRemoteException.create(handle.getError()));
+          throw Status.INTERNAL.withDescription(handle.getError().getMessage()).withCause(UserRemoteException.create(handle.getError())).asRuntimeException();
         }
+        logger.debug("prepared statement handle for job id {} has succeeded", ExternalIdHelper.toQueryId(externalId));
         PreparedStatement statement = handle.getPreparedStatement();
         Ticket ticket = new Ticket(statement.getServerHandle().toByteArray());
         FlightEndpoint endpoint = new FlightEndpoint(ticket, location);
-        FlightInfo info = new FlightInfo(fromMetadata(statement.getColumnsList()), descriptor, Lists.newArrayList(endpoint), -1L, -1L);
+        logger.debug("flight endpoint for job id {} has been created with ticket {}", ExternalIdHelper.toQueryId(externalId), new String(ticket.getBytes()));
+        Schema schema = fromMetadata(statement.getColumnsList());
+        FlightInfo info = new FlightInfo(schema, descriptor, Lists.newArrayList(endpoint), -1L, -1L);
+        logger.debug("flight info for job id {} has been created with schema {}", ExternalIdHelper.toQueryId(externalId), schema.toJson());
         return info;
-      } catch (ExecutionException e) {
-        throw new RuntimeException(e.getCause());
-      } catch (InterruptedException e) {
-        throw new RuntimeException(e);
+      } catch (Exception e) {
+        logger.warn("prepared statement handle for job id " + ExternalIdHelper.toQueryId(externalId) + " has failed", UserException.parseError(e).buildSilently());
+        throw Status.UNKNOWN.withCause(UserException.parseError(e).buildSilently()).asRuntimeException();
       }
     }
 
     private Schema fromMetadata(List<ResultColumnMetadata> rcmd) {
 
       Schema schema = new Schema(rcmd.stream().map(md -> {
-        ArrowType arrowType = SqlTypeNameToArrowType.toArrowType(md.getDataType());
+        ArrowType arrowType = SqlTypeNameToArrowType.toArrowType(md);
         FieldType fieldType = new FieldType(md.getIsNullable(), arrowType, null, null);
         return new Field(md.getColumnName(), fieldType, null);
       }).collect(Collectors.toList()));
@@ -188,7 +198,7 @@ private Schema fromMetadata(List<ResultColumnMetadata> rcmd) {
 
     @Override
     public void sendData(RpcOutcomeListener<Ack> outcomeListener, QueryWritableBatch result) {
-      throw new IllegalStateException();
+      throw Status.UNIMPLEMENTED.asRuntimeException();
     }
 
     @Override

src/main/java/com/dremio/flight/SqlTypeNameToArrowType.java

@@ -23,16 +23,17 @@
 import org.apache.arrow.vector.types.pojo.ArrowType.Binary;
 import org.apache.arrow.vector.types.pojo.ArrowType.Bool;
 import org.apache.arrow.vector.types.pojo.ArrowType.Date;
+import org.apache.arrow.vector.types.pojo.ArrowType.Decimal;
 import org.apache.arrow.vector.types.pojo.ArrowType.FloatingPoint;
 import org.apache.arrow.vector.types.pojo.ArrowType.Int;
 import org.apache.arrow.vector.types.pojo.ArrowType.Interval;
-import org.apache.arrow.vector.types.pojo.ArrowType.List;
 import org.apache.arrow.vector.types.pojo.ArrowType.Null;
-import org.apache.arrow.vector.types.pojo.ArrowType.Struct;
 import org.apache.arrow.vector.types.pojo.ArrowType.Time;
 import org.apache.arrow.vector.types.pojo.ArrowType.Timestamp;
 import org.apache.arrow.vector.types.pojo.ArrowType.Utf8;
 
+import com.dremio.exec.proto.UserProtos;
+
 /***
  * Gets Dremio RPC-/protobuf-level data type for given SQL data type name.
  * returns the canonical keyword sequence for SQL data type (leading keywords in
@@ -41,14 +42,17 @@
  */
 public class SqlTypeNameToArrowType {
 
-  public static ArrowType toArrowType(String typeName) {
+  public static ArrowType toArrowType(UserProtos.ResultColumnMetadata type) {
+    String typeName = type.getDataType();
     switch (typeName) {
       case "NULL":
         return new Null();
       case "MAP":
-        return new Struct(); //todo inner type?
+        throw new UnsupportedOperationException("have not implemented map");
+        //return new Struct(); //todo inner type?
       case "ARRAY":
-        return new List(); //todo inner type?
+        throw new UnsupportedOperationException("have not implemented array");
+        //return new List(); //todo inner type?
       case "UNION":
         throw new UnsupportedOperationException("have not implemented unions");
         //return new Union(); //todo inner type?
@@ -71,7 +75,7 @@ public static ArrowType toArrowType(String typeName) {
       case "BOOLEAN":
         return new Bool();
       case "DECIMAL":
-        throw new UnsupportedOperationException("have not implemented decimal");
+        return new Decimal(type.getPrecision(), type.getScale());
       case "DATE":
         return new Date(DateUnit.MILLISECOND);
       case "TIME":