private void readData () throws SQLException {

      // Close current cursor and reaopen.
      int currentFetchSize = getFetchSize();
      setFetchSize(0);
      close();
      setFetchSize(currentFetchSize);
      moveToInsertRow();

      Document document;

      // load fetch size documents into the rowset
      for (int i = 1; i <= getFetchSize(); i++) {

         try {

            if (this.cursor.hasNext()) {
               document = this.cursor.next();

               // if the sink database is mongodb, the document will be inserted as is
               if (Boolean.TRUE.equals(this.isSourceAndSinkMongo)) {
                  updateObject(1, document);
               } else {

                  for (int j = 0; j <= getMetaData().getColumnCount() - 1; j++) {
                     String columnName = getMetaData().getColumnName(j + 1);
                     int columnType = getMetaData().getColumnType(j + 1);

                     switch (columnType) {
                        case Types.VARCHAR:
                        case Types.CHAR:
                        case Types.LONGVARCHAR:
                           if (document.getString(columnName) == null) updateNull(j + 1);
                           else updateString(j + 1, document.getString(columnName));
                           break;
                        case Types.INTEGER:
                        case Types.TINYINT:
                        case Types.SMALLINT:
                           if (document.getInteger(columnName) == null) updateNull(j + 1);
                           else updateInt(j + 1, document.getInteger(columnName));
                           break;
                        case Types.BIGINT:
                        case Types.NUMERIC:
                        case Types.DECIMAL:
                           if (document.getLong(columnName) == null) updateNull(j + 1);
                           else updateBigDecimal(j + 1, BigDecimal.valueOf(document.getLong(columnName)));
                           break;
                        case Types.DOUBLE:
                           if (document.getDouble(columnName) == null) updateNull(j + 1);
                           else updateDouble(j + 1, document.getDouble(columnName));
                           break;
                        case Types.TIMESTAMP_WITH_TIMEZONE:
                           if (document.getDate(columnName) == null) updateNull(j + 1);
                              // convert to offsetDateTime
                           else updateObject(j + 1, document.getDate(columnName).toInstant().atOffset(ZoneOffset.UTC));
                           break;
                        case Types.BINARY:
                        case Types.BLOB:
                           Binary bin = document.get(columnName, org.bson.types.Binary.class);
                           if (bin == null) updateNull(j + 1);
                           else updateBytes(j + 1, bin.getData());
                           break;
                        case Types.BOOLEAN:
                           if (document.getBoolean(columnName) == null) updateNull(j + 1);
                           else updateBoolean(j + 1, document.getBoolean(columnName));
                           break;
                        case Types.ROWID:
                           ObjectId oid = document.getObjectId(columnName);
                           if (oid == null) updateNull(j + 1);
                           else updateString(j + 1, oid.toString());
                           break;
                        case Types.ARRAY:
                           List list = document.get(columnName, List.class);
                           String jsonArr = new Gson().toJson(list);
                           if (list == null) updateNull(j + 1);
                           else updateString(j + 1, jsonArr);
                           break;
                        default:
                           String json = BsonUtils.toJson(document.get(columnName, org.bson.Document.class));
                           if (json == null) updateNull(j + 1);
                           else updateString(j + 1, json);
                           break;
                     }
                  }
               }
               insertRow();
               incrementRowCount();
            }
         } catch (Exception e) {
            LOG.error("MongoDB error: {}", e.getMessage(), e);
            throw e;
         }
      }

      moveToCurrentRow();
      beforeFirst();
   }
Maps a given class type to a corresponding SQL type.

@param typeString the class type as a string
@return the corresponding SQL type, or {@link java.sql.Types#OTHER} if no corresponding type is found.
src/main/java/org/replicadb/rowset/MongoDBRowSetImpl.java