DLPX-93245 HOP : Migrating Existing Kettle Patches related to DB to the forked local HOP project repo (#10)

* DLPX-93245 HOP : Migrating Existing Kettle Patches related to DB to the forked local HOP project repo

PR URL: https://www.github.com/delphix/hop/pull/10

Author: dx-atulyadas

core/src/main/java/org/apache/hop/core/database/Database.java

@@ -66,6 +66,8 @@ public class Database implements IVariables, ILoggingObject {
 
   private int rowlimit;
   private int commitsize;
+  private boolean delphixSpecial = false;
+  private int fetchSize = 0;
 
   private Connection connection;
 
@@ -587,6 +589,23 @@ public void cancelStatement(Statement statement) throws HopDatabaseException {
    * @param commsize The number of rows to wait before doing a commit on the connection.
    */
   public void setCommit(int commsize) {
+    /*
+     * We override this method to set fetch size in the case where it was called
+     * previously with a "magic" value. It would be much cleaner to just add a
+     * setFetchSize() method, but the way we build dms-core-gate, changes here
+     * aren't visible to link against when building our code.
+     */
+    if (commsize == Integer.MIN_VALUE + 42) {
+      delphixSpecial = true;
+      return;
+    }
+    if (delphixSpecial) {
+      fetchSize = commsize;
+      delphixSpecial = false;
+      return;
+    }
+    // End Delphix extra-special case.
+
     commitsize = commsize;
     String onOff = (commitsize <= 0 ? "on" : "off");
     try {
@@ -1041,7 +1060,7 @@ && getDatabaseMetaData().supportsBatchUpdates()
           && databaseMeta.supportsBatchUpdates()
           && Utils.isEmpty(connectionGroup);
     } catch (SQLException e) {
-      throw createHopDatabaseBatchException("Error determining whether to use batch", e);
+      throw createHopDatabaseBatchException("Error determining whether to use batch", e ,this.log.getLogLevel());
     }
   }
 
@@ -1108,10 +1127,10 @@ public boolean insertRow(PreparedStatement ps, boolean batch, boolean handleComm
 
       return rowsAreSafe;
     } catch (BatchUpdateException ex) {
-      throw createHopDatabaseBatchException("Error updating batch", ex);
+      throw createHopDatabaseBatchException("Error updating batch", ex, this.log.getLogLevel());
     } catch (SQLException ex) {
       if (isBatchUpdate) {
-        throw createHopDatabaseBatchException("Error updating batch", ex);
+        throw createHopDatabaseBatchException("Error updating batch", ex, this.log.getLogLevel());
       } else {
         throw new HopDatabaseException("Error inserting/updating row", ex);
       }
@@ -1219,9 +1238,13 @@ public void emptyAndCommit(
     }
   }
 
+  public static HopDatabaseBatchException createHopDatabaseBatchException( String message, SQLException ex) {
+    return createHopDatabaseBatchException(message, ex, DefaultLogLevel.getLogLevel());
+  }
+
   public static HopDatabaseBatchException createHopDatabaseBatchException(
-      String message, SQLException ex) {
-    HopDatabaseBatchException kdbe = new HopDatabaseBatchException(message, ex);
+      String message, SQLException ex, LogLevel logLevel1) {
+    HopDatabaseBatchException kdbe = new HopDatabaseBatchException(message, ex, logLevel1);
     if (ex instanceof BatchUpdateException) {
       kdbe.setUpdateCounts(((BatchUpdateException) ex).getUpdateCounts());
     } else {
@@ -1465,13 +1488,13 @@ public ResultSet openQuery(
 
         if (canWeSetFetchSize(pstmt)) {
           int maxRows = pstmt.getMaxRows();
-          int fs = Const.FETCH_SIZE <= maxRows ? maxRows : Const.FETCH_SIZE;
+          int fs = Math.max(maxRows, fetchSize > 0 ? fetchSize : Const.FETCH_SIZE);
           if (databaseMeta.isMySqlVariant()) {
             setMysqlFetchSize(pstmt, fs, maxRows);
           } else {
             pstmt.setFetchSize(fs);
           }
-
+          log.logBasic("Statement fetch size set to " + fs);
           pstmt.setFetchDirection(fetchMode);
         }
 
@@ -1487,13 +1510,14 @@ public ResultSet openQuery(
         selStmt = connection.createStatement();
         log.snap(Metrics.METRIC_DATABASE_CREATE_SQL_STOP, databaseMeta.getName());
         if (canWeSetFetchSize(selStmt)) {
-          int fs =
-              Const.FETCH_SIZE <= selStmt.getMaxRows() ? selStmt.getMaxRows() : Const.FETCH_SIZE;
+          int fs = Math.max(selStmt.getMaxRows(), fetchSize > 0 ? fetchSize : Const.FETCH_SIZE);
           if (databaseMeta.getIDatabase().isMySqlVariant() && databaseMeta.isStreamingResults()) {
             selStmt.setFetchSize(Integer.MIN_VALUE);
           } else {
             selStmt.setFetchSize(fs);
           }
+          log.logBasic("Statement fetch size set to " + fs);
+
           selStmt.setFetchDirection(fetchMode);
         }
         if (rowlimit > 0 && databaseMeta.supportsSetMaxRows()) {
@@ -1523,10 +1547,16 @@ public ResultSet openQuery(
   }
 
   private boolean canWeSetFetchSize(Statement statement) throws SQLException {
-    return databaseMeta.isFetchSizeSupported()
-        && (statement.getMaxRows() > 0
+    if (!databaseMeta.isFetchSizeSupported()) {
+      return false;
+    }
+    // Override for delphix
+    if (fetchSize > 0) {
+      return true;
+    }
+    return statement.getMaxRows() > 0
             || databaseMeta.getIDatabase().isPostgresVariant()
-            || (databaseMeta.isMySqlVariant() && databaseMeta.isStreamingResults()));
+            || ( databaseMeta.isMySqlVariant() && databaseMeta.isStreamingResults() );
   }
 
   public ResultSet openQuery(PreparedStatement ps, IRowMeta params, Object[] data)
@@ -1541,24 +1571,25 @@ public ResultSet openQuery(PreparedStatement ps, IRowMeta params, Object[] data)
       setValues(params, data, ps); // set the parameters!
       log.snap(Metrics.METRIC_DATABASE_SQL_VALUES_STOP, databaseMeta.getName());
 
+      if ( rowlimit > 0 && databaseMeta.supportsSetMaxRows() ) {
+        ps.setMaxRows( rowlimit );
+      }
+
       if (canWeSetFetchSize(ps)) {
         int maxRows = ps.getMaxRows();
-        int fs = Const.FETCH_SIZE <= maxRows ? maxRows : Const.FETCH_SIZE;
+        int fs = Math.max(maxRows, fetchSize > 0 ? fetchSize : Const.FETCH_SIZE);
         // mysql have some restriction on fetch size assignment
         if (databaseMeta.isMySqlVariant()) {
           setMysqlFetchSize(ps, fs, maxRows);
         } else {
           // other databases seems not.
           ps.setFetchSize(fs);
         }
+        log.logBasic("Statement fetch size set to " + fs);
 
         ps.setFetchDirection(ResultSet.FETCH_FORWARD);
       }
 
-      if (rowlimit > 0 && databaseMeta.supportsSetMaxRows()) {
-        ps.setMaxRows(rowlimit);
-      }
-
       log.snap(Metrics.METRIC_DATABASE_EXECUTE_SQL_START, databaseMeta.getName());
       res = ps.executeQuery();
       log.snap(Metrics.METRIC_DATABASE_EXECUTE_SQL_STOP, databaseMeta.getName());

core/src/main/java/org/apache/hop/core/exception/HopDatabaseBatchException.java

@@ -17,8 +17,14 @@
 
 package org.apache.hop.core.exception;
 
+import java.sql.SQLException;
 import java.util.List;
 
+import org.apache.hop.core.Const;
+import org.apache.hop.core.logging.DefaultLogLevel;
+import org.apache.hop.core.logging.LogLevel;
+
+
 /** This exception is used by the Database class. */
 public class HopDatabaseBatchException extends HopDatabaseException {
   public static final long serialVersionUID = 0x8D8EA0264F7A1C0EL;
@@ -27,6 +33,8 @@ public class HopDatabaseBatchException extends HopDatabaseException {
 
   private List<Exception> exceptionsList;
 
+  private LogLevel logLevel = DefaultLogLevel.getLogLevel();
+
   /** Constructs a new throwable with null as its detail message. */
   public HopDatabaseBatchException() {
     super();
@@ -65,6 +73,20 @@ public HopDatabaseBatchException(String message, Throwable cause) {
     super(message, cause);
   }
 
+  /**
+   * Constructs a new throwable with the specified detail message and cause.
+   *
+   * @param message
+   *          the detail message (which is saved for later retrieval by the getMessage() method).
+   * @param cause
+   *          the cause (which is saved for later retrieval by the getCause() method). (A null value is permitted, and
+   *          indicates that the cause is nonexistent or unknown.)
+   */
+  public HopDatabaseBatchException( String message, Throwable cause, LogLevel logLevel ) {
+    super( message, cause );
+    this.logLevel = logLevel;
+  }
+
   /** @return Returns the updateCounts. */
   public int[] getUpdateCounts() {
     return updateCounts;
@@ -82,4 +104,42 @@ public void setExceptionsList(List<Exception> exceptionsList) {
   public List<Exception> getExceptionsList() {
     return exceptionsList;
   }
+
+  /*
+   * This method exists to fix DLPX-56268 (Masking does not see Hop getNextException to surface the cause of a SQLException).
+   *
+   * The HopException class overrides this method of Exception class to get the root cause. This
+   * class is a subclass of HopException class. It has an additional member field to store list
+   * of exceptions. The static method createHopDatabaseBatchException in Database class is used
+   * to create an instance of this class. In that method all the exceptions associated with SQLException
+   * are retrieved using the getNextException and stored in this list.
+   *
+   * While logging the error message, the getMessage method is used. Since it is a subclass of HopException class, it
+   * uses the method defined there. That definition ignores this list of exceptions. Thus these messages are supressed.
+   *
+   * We fix this by overriding the getMessage method here and log all the exceptions present in the exceptionList variable.
+   * It also calls the HopException's getMessage method so that the previous behavior is maintained while adding
+   * additional log messages.
+   */
+  @Override
+  public String getMessage(){
+    String retval = Const.CR;
+    retval += super.getMessage() + Const.CR;
+
+    for(Exception exc: exceptionsList){
+      if(exc instanceof SQLException){
+        SQLException sqlException = (SQLException)exc;
+        retval += "Next Exception: SQLState( "
+                + sqlException.getSQLState() + ") ErrorCode("
+                + sqlException.getErrorCode() + ")";
+        if(this.logLevel.isDetailed()){
+          retval += " Message: " + sqlException.getMessage();
+        }
+        retval += Const.CR;
+      } else {
+        retval += "Next Exception: " + exc.getClass() + Const.CR;
+      }
+    }
+    return retval;
+  }
 }

core/src/main/java/org/apache/hop/core/row/value/ValueMetaBase.java

@@ -168,6 +168,10 @@ public class ValueMetaBase implements IValueMeta {
   protected int originalNullable;
   protected boolean originalSigned;
 
+  // Added for DLPX-46230 / DLPX-82789 to prevent using setBlob on non-blob columns in Oracle
+  protected boolean isLargeObject = false;
+
+
   protected boolean ignoreWhitespace;
 
   protected final Comparator<Object> comparator;
@@ -4118,11 +4122,7 @@ public Object convertDataFromString(
     boolean isStringValue = outValueType == IValueMeta.TYPE_STRING;
     Object emptyValue = isStringValue ? Const.NULL_STRING : null;
 
-    boolean isEmptyAndNullDiffer =
-        convertStringToBoolean(
-            Const.NVL(System.getProperty(Const.HOP_EMPTY_STRING_DIFFERS_FROM_NULL, "N"), "N"));
-
-    if (pol == null && isStringValue && isEmptyAndNullDiffer) {
+    if ( pol == null && isStringValue && emptyStringAndNullAreDifferent ) {
       pol = Const.NULL_STRING;
     }
 
@@ -4681,6 +4681,7 @@ public IValueMeta getValueFromSqlType(
       int precision = -1;
       int valtype = IValueMeta.TYPE_NONE;
       boolean isClob = false;
+      isLargeObject = false;
 
       int type = rm.getColumnType(index);
       boolean signed = false;
@@ -5116,6 +5117,32 @@ public IValueMeta getMetadataPreview(
             }
           }
 
+          /*
+           * This logic exists to fix DLPX-51415 (Sybase - Numeric Conversion Error).
+           *
+           * The logic above will typically force DECIMAL/NUMERIC types with a scale ('precision' in Kettle terms) > 0
+           * to be cast to Java Doubles. A Double is a binary type, so it can not exactly represent arbitrary decimal
+           * values.
+           *
+           * This is particularly problematic for Sybase, because Sybase has some inconsistencies related to how it
+           * deals with inserts/updates of values with greater precision than can be held by type of the column, for
+           * instance inserting a value 1.2345 into a column with a type DECIMAL(4,2). For certain combinations of
+           * jTDS/jConnect and batched/non-batched mode, Sybase will throw an error of the form
+           *
+           *      Scale error during implicit conversion of NUMERIC value '1.2345' to a DECIMAL field.
+           *
+           * This causes a problem when the inexact floating point values read from Sybase columns are inserted back
+           * into Sybase NUMERIC/DECIMAL columns. We fix this by reading NUMERIC/DECIMAL values into a Java decimal
+           * type which can store the exact value correctly. We will likely want this behavior for all platforms
+           * eventually, but for now we restrict this behavior to Sybase, in order to minimize the change, and
+           * thereby the risk associated with the patch.
+           */
+          if (databaseMeta.getIDatabase().isSybaseVariant()
+                  && (type == Types.NUMERIC || type == Types.DECIMAL)
+                  && precision > 0) {
+            valtype = IValueMeta.TYPE_BIGNUMBER;
+          }
+
           break;
 
         case Types.TIMESTAMP:
@@ -5153,6 +5180,8 @@ public IValueMeta getMetadataPreview(
 
         case Types.BINARY:
         case Types.BLOB:
+          isLargeObject = true;
+          // fallthrough
         case Types.VARBINARY:
         case Types.LONGVARBINARY:
           valtype = IValueMeta.TYPE_BINARY;
@@ -5167,9 +5196,10 @@ public IValueMeta getMetadataPreview(
           } else if ((databaseMeta.getIDatabase().isOracleVariant())
               && (originalColumnType == Types.VARBINARY
                   || originalColumnType == Types.LONGVARBINARY)) {
-            // set the length for Oracle "RAW" or "LONGRAW" data types
-            valtype = IValueMeta.TYPE_STRING;
-            length = originalColumnDisplaySize;
+            // set the length for Oracle "RAW" or "LONGRAW" data
+            // For DLPX-82789, these type are fundamentally binary and should be treated as such
+            // valtype = IValueMeta.TYPE_STRING;
+            // length = originalColumnDisplaySize;
           } else if (databaseMeta.isMySqlVariant()
               && (originalColumnType == Types.VARBINARY
                   || originalColumnType == Types.LONGVARBINARY)) {
@@ -5246,7 +5276,7 @@ public Object getValueFromResultSet(IDatabase iDatabase, ResultSet resultSet, in
           }
           break;
         case IValueMeta.TYPE_BINARY:
-          if (iDatabase.supportsGetBlob()) {
+          if ( isLargeObject && iDatabase.supportsGetBlob()) {
             Blob blob = resultSet.getBlob(index + 1);
             if (blob != null) {
               data = blob.getBytes(1L, (int) blob.length());
@@ -5314,19 +5344,23 @@ public void setPreparedStatementValue(
           }
           break;
         case IValueMeta.TYPE_INTEGER:
-          if (!isNull(data)) {
-            if (databaseMeta.supportsSetLong()) {
+          if (databaseMeta.supportsSetLong()) {
+            if (!isNull(data)) {
               preparedStatement.setLong(index, getInteger(data).longValue());
             } else {
+              preparedStatement.setNull( index, Types.BIGINT );
+            }
+          } else {
+            if (!isNull(data)) {
               double d = getNumber(data).doubleValue();
               if (databaseMeta.supportsFloatRoundingOnUpdate() && getPrecision() >= 0) {
                 preparedStatement.setDouble(index, d);
               } else {
                 preparedStatement.setDouble(index, Const.round(d, getPrecision()));
               }
+            } else {
+              preparedStatement.setNull(index, Types.DOUBLE);
             }
-          } else {
-            preparedStatement.setNull(index, Types.INTEGER);
           }
           break;
         case IValueMeta.TYPE_STRING:

core/src/test/java/org/apache/hop/core/row/value/ValueMetaBaseTest.java

@@ -318,8 +318,8 @@ public void testConvertDataFromStringToString() throws HopValueException {
             inputValueNullString, inValueMetaString, nullIf, ifNull, trimType);
     assertEquals(
         "HOP_EMPTY_STRING_DIFFERS_FROM_NULL = Y: "
-            + "Conversion from null string must return empty string",
-        StringUtils.EMPTY,
+            + "Conversion from null string must return null",
+        null,
         result);
   }