JdbcDao support including status column in Query Automatic,Stream read and write support five Serialize solution

Author: robinhood-jim

common/src/main/java/com/robin/core/fileaccess/iterator/AbstractResIterator.java

@@ -1,5 +1,7 @@
 package com.robin.core.fileaccess.iterator;
 
+import com.google.gson.Gson;
+import com.robin.comm.util.json.GsonUtil;
 import com.robin.core.base.exception.OperationNotSupportException;
 import com.robin.core.fileaccess.fs.AbstractFileSystemAccessor;
 import com.robin.core.fileaccess.meta.DataCollectionMeta;
@@ -22,6 +24,7 @@ public abstract class AbstractResIterator implements IResourceIterator {
     protected Map<String, DataSetColumnMeta> columnMap=new HashMap<>();
     protected Logger logger= LoggerFactory.getLogger(getClass());
     protected String identifier;
+    protected Gson gson= GsonUtil.getGson();
     protected AbstractResIterator(){
 
     }

common/src/main/java/com/robin/core/fileaccess/writer/AbstractQueueWriter.java

@@ -25,7 +25,7 @@ public abstract class AbstractResourceWriter implements IResourceWriter{
     protected Map<String, Void> columnMap=new HashMap<String, Void>();
     protected List<String> columnList=new ArrayList<String>();
     protected Logger logger= LoggerFactory.getLogger(getClass());
-    protected String valueType= ResourceConst.VALUE_TYPE.JSON.getValue();
+    protected String valueType= ResourceConst.SERIALIZETYPE.JSON.getValue();
     protected Schema schema;
     protected Map<String,Object> cfgMap;
     protected Gson gson= GsonUtil.getGson();

common/src/main/java/com/robin/core/fileaccess/writer/AbstractResourceWriter.java

@@ -28,4 +28,6 @@
 	String jdbcDao() default "jdbcDao";
 	//Does all field must declare explicit?
 	boolean explicit() default false;
+	String statusColumn() default "";
+	String statusValue() default "";
 }

common/src/main/java/com/robin/core/resaccess/iterator/AbstractQueueIterator.java

@@ -322,13 +322,19 @@ public <T extends BaseObject> List<T> queryByField(Class<T> type, String fieldNa
             StringBuilder buffer = new StringBuilder();
             buffer.append(getWholeSelectSql(type)).append(Const.SQL_WHERE);
             StringBuilder queryBuffer = new StringBuilder();
+            AnnotationRetriever.EntityContent<T> tableDef = AnnotationRetriever.getMappingTableByCache(type);
             Map<String, FieldContent> map1 = AnnotationRetriever.getMappingFieldsMapCache(type);
             List<FieldContent> fields = AnnotationRetriever.getMappingFieldsCache(type);
             List<Map<String, Object>> rsList;
             if (map1.containsKey(fieldName)) {
+                List<Object> queryParam=Lists.newArrayList(fieldValues);
                 generateQuerySqlBySingleFields(map1.get(fieldName), oper, queryBuffer, fieldValues.length);
+                if(tableDef.isContainStatusColumn()){
+                    appendStatusColumn(queryBuffer,tableDef.getStatusColumn());
+                    queryParam.add(tableDef.getStatusValue());
+                }
                 buffer.append(queryBuffer);
-                rsList = queryBySql(buffer.toString(), fieldValues);
+                rsList = queryBySql(buffer.toString(), queryParam.toArray());
                 wrapList(type, retlist, fields, rsList);
             } else {
                 throw new DAOException("query Field not in entity");
@@ -374,19 +380,25 @@ public <T extends BaseObject> List<T> queryByFieldOrderBy(Class<T> type, String
         List<T> retlist = new ArrayList<>();
         try {
             StringBuilder builder = new StringBuilder();
+            AnnotationRetriever.EntityContent<T> tableDef = AnnotationRetriever.getMappingTableByCache(type);
             List<FieldContent> fields = AnnotationRetriever.getMappingFieldsCache(type);
             builder.append(getWholeSelectSql(type)).append(Const.SQL_WHERE);
             StringBuilder queryBuffer = new StringBuilder();
             Map<String, FieldContent> map1 = AnnotationRetriever.getMappingFieldsMapCache(type);
 
             List<Map<String, Object>> rsList;
             if (map1.containsKey(fieldName)) {
+                List<Object> queryParam=Lists.newArrayList(fieldValues);
                 generateQuerySqlBySingleFields(map1.get(fieldName), oper, queryBuffer, fieldValues.length);
+                if(tableDef.isContainStatusColumn()){
+                    appendStatusColumn(queryBuffer,tableDef.getStatusColumn());
+                    queryParam.add(tableDef.getStatusValue());
+                }
                 builder.append(queryBuffer);
                 if (!ObjectUtils.isEmpty(orderByStr)) {
                     builder.append(" order by ").append(orderByStr);
                 }
-                rsList = queryBySql(builder.toString(), fieldValues);
+                rsList = queryBySql(builder.toString(), queryParam.toArray());
                 wrapList(type, retlist, fields, rsList);
             } else {
                 throw new DAOException("query Field not in entity");
@@ -942,6 +954,9 @@ private void generateQuerySqlBySingleFields(FieldContent columncfg, Const.OPERAT
                 break;
         }
     }
+    private void appendStatusColumn(StringBuilder queryBuilder,String statusColumn){
+        queryBuilder.append(" AND ").append(statusColumn).append("=?");
+    }
 
     private void wrapResultToModelWithKey(BaseObject obj, Map<String, Object> map, List<FieldContent> fields, Serializable pkObj) throws Throwable {
         for (FieldContent field : fields) {

core/src/main/java/com/robin/core/base/annotation/MappingEntity.java

@@ -212,6 +212,13 @@ private static <T extends BaseObject> EntityContent<T> getMappingTableEntity(Cla
             if (!ObjectUtils.isEmpty(jdbcDao)) {
                 content.setJdbcDao(jdbcDao);
             }
+            if(!ObjectUtils.isEmpty(entity.statusColumn())){
+                content.setStatusColumn(entity.statusColumn());
+                content.setContainStatusColumn(true);
+            }
+            if(!ObjectUtils.isEmpty(entity.statusValue())){
+                content.setStatusValue(entity.statusValue());
+            }
         } else {
             flag = clazz.isAnnotationPresent(Entity.class);
             if (flag) {
@@ -749,6 +756,9 @@ public static class EntityContent<T extends BaseObject> {
         private String schema;
         private String jdbcDao;
         private Class<T> clazz;
+        private boolean containStatusColumn=false;
+        private String statusColumn;
+        private String statusValue=Const.VALID;
 
         public EntityContent(String tableName) {
             this.tableName = tableName;

core/src/main/java/com/robin/core/base/dao/JdbcDao.java

@@ -356,7 +356,10 @@ public static <T extends BaseObject> SelectSegment getSelectPkSegment(Class<T> c
                 segment.getAvailableFields().add(field);
                 sqlbuffer.append(field.getFieldName()).append(Const.SQL_AS).append(field.getPropertyName()).append(",");
             }
-
+        }
+        if(tableDef.isContainStatusColumn()){
+            wherebuffer.append(tableDef.getStatusColumn()).append("=? and ");
+            selectObjs.add(tableDef.getStatusValue());
         }
         sqlbuffer.deleteCharAt(sqlbuffer.length() - 1).append(Const.SQL_FROM);
         appendSchemaAndTable(tableDef, sqlbuffer, sqlGen);

core/src/main/java/com/robin/core/base/dao/util/AnnotationRetriever.java

@@ -1,6 +1,9 @@
 package com.robin.core.base.util;
 
 
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+
 public class ResourceConst {
     public static final String WORKINGPATHPARAM="output.workingPath";
     public static final String USETMPFILETAG="output.usingTmpFiles";
@@ -20,6 +23,7 @@ public class ResourceConst {
     public static final String STORAGEFILTERSQL="storage.FilterSql";
     public static final String PARQUETFILEFORMAT="parquet.file.format";
     public static final String USEADMINTG="useAdmin";
+    public static final String SERIALIZETYPE_COLUMN="output.SerializeType";
 
     public enum IngestType {
         TYPE_HDFS(1L,"HDFS"),
@@ -93,22 +97,29 @@ public String toString() {
             return String.valueOf(this.value);
         }
     }
-    public enum VALUE_TYPE{
+    public enum SERIALIZETYPE {
         AVRO("avro"),
         JSON("json"),
-        XML("xml"),
         PROTOBUF("proto"),
-        ORC("orc"),
-        CSV("csv"),
-        PARQUET("parquet");
+        MESSAGEPACK("msgpack"),
+        BIJECTION("bijection");
         private String value;
-        VALUE_TYPE(String value){
+        SERIALIZETYPE(String value){
             this.value=value;
         }
 
         public String getValue() {
             return value;
         }
+        public static SERIALIZETYPE ParseFrom(String input){
+            Assert.isTrue(!ObjectUtils.isEmpty(input),"");
+            for(SERIALIZETYPE type :values()){
+                if(type.getValue().equalsIgnoreCase(input)){
+                    return type;
+                }
+            }
+            throw new IllegalArgumentException("unknown valueType"+input);
+        }
     }
     public enum RESTYPE{
         DIR("1"),

core/src/main/java/com/robin/core/base/dao/util/EntityMappingUtil.java

@@ -126,6 +126,9 @@ public boolean isQueueType(){
 	}
 	public static class Builder{
 		private final DataCollectionMeta meta=new DataCollectionMeta();
+		public static Builder newBuilder(){
+			return  new Builder();
+		}
 		public Builder(){
 
 		}