fix msssql tests take 3

airbytehq · Feb 24, 2024 · 2503621 · 2503621
1 parent 2b92c25
commit 2503621
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 11 deletions.
diff --git a/...cdk/core/src/main/java/io/airbyte/cdk/db/jdbc/AbstractJdbcCompatibleSourceOperations.java b/...cdk/core/src/main/java/io/airbyte/cdk/db/jdbc/AbstractJdbcCompatibleSourceOperations.java
@@ -30,12 +30,15 @@
 import java.time.format.DateTimeParseException;
 import java.util.Base64;
 import java.util.Collections;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Source operation skeleton for JDBC compatible databases.
  */
 public abstract class AbstractJdbcCompatibleSourceOperations<Datatype> implements JdbcCompatibleSourceOperations<Datatype> {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(AbstractJdbcCompatibleSourceOperations.class);
   /**
    * A Date representing the earliest date in CE. Any date before this is in BCE.
    */
@@ -107,7 +110,9 @@ protected void putBigInt(final ObjectNode node, final String columnName, final R
   }
 
   protected void putDouble(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException {
-    node.put(columnName, DataTypeUtils.returnNullIfInvalid(() -> resultSet.getDouble(index), Double::isFinite));
+    double val = DataTypeUtils.returnNullIfInvalid(() -> resultSet.getDouble(index), Double::isFinite);
+    LOGGER.info("SGXputDouble " + columnName + " = " + val);
+    node.put(columnName, val);
   }
 
   protected void putFloat(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException {

diff --git a/.../java/io/airbyte/cdk/integrations/standardtest/source/AbstractSourceDatabaseTypeTest.java b/.../java/io/airbyte/cdk/integrations/standardtest/source/AbstractSourceDatabaseTypeTest.java
@@ -185,24 +185,22 @@ public UnexpectedRecord(String streamName, String unexpectedValue) {
       }
     }
 
-    assertTrue(unexpectedValues.isEmpty(),
-        unexpectedValues.stream().map((entry) -> // stream each entry, map it to string value
-        "The stream '" + entry.streamName + "' checking type '" + testByName.get(entry.streamName).getSourceType() + "' initialized at "
-            + testByName.get(entry.streamName).getDeclarationLocation() + " got unexpected values: " + entry.unexpectedValue)
-            .collect(Collectors.joining("\n"))); // and join them
-
     // Gather all the missing values, so we don't stop the test in the first missed one
     expectedValues.forEach((streamName, values) -> {
       if (!values.isEmpty()) {
         missedValues.add(new MissedRecords(streamName, values));
       }
     });
 
-    assertTrue(missedValues.isEmpty(),
+    assertTrue(missedValues.isEmpty() && unexpectedValues.isEmpty(),
         missedValues.stream().map((entry) -> // stream each entry, map it to string value
         "The stream '" + entry.streamName + "' checking type '" + testByName.get(entry.streamName).getSourceType() + "' initialized at "
             + testByName.get(entry.streamName).getDeclarationLocation() + " is missing values: " + entry.missedValues)
-            .collect(Collectors.joining("\n"))); // and join them
+            .collect(Collectors.joining("\n")) +
+            unexpectedValues.stream().map((entry) -> // stream each entry, map it to string value
+            "The stream '" + entry.streamName + "' checking type '" + testByName.get(entry.streamName).getSourceType() + "' initialized at "
+                + testByName.get(entry.streamName).getDeclarationLocation() + " got unexpected values: " + entry.unexpectedValue)
+                .collect(Collectors.joining("\n"))); // and join them
   }
 
   protected String getValueFromJsonNode(final JsonNode jsonNode) throws IOException {

diff --git a/...ource-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlSourceOperations.java b/...ource-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlSourceOperations.java
@@ -16,6 +16,7 @@
 import com.microsoft.sqlserver.jdbc.Geography;
 import com.microsoft.sqlserver.jdbc.Geometry;
 import com.microsoft.sqlserver.jdbc.SQLServerResultSetMetaData;
+import io.airbyte.cdk.db.DataTypeUtils;
 import io.airbyte.cdk.db.jdbc.JdbcSourceOperations;
 import io.airbyte.protocol.models.JsonSchemaType;
 import java.sql.JDBCType;
@@ -69,13 +70,13 @@ private void putValue(final JDBCType columnType,
                         final int colIndex,
                         final ObjectNode json)
       throws SQLException {
+    LOGGER.info("SGX columnType= " + columnType);
     switch (columnType) {
       case BIT, BOOLEAN -> putBoolean(json, columnName, resultSet, colIndex);
       case TINYINT, SMALLINT -> putShortInt(json, columnName, resultSet, colIndex);
       case INTEGER -> putInteger(json, columnName, resultSet, colIndex);
       case BIGINT -> putBigInt(json, columnName, resultSet, colIndex);
-      case FLOAT, DOUBLE -> putDouble(json, columnName, resultSet, colIndex);
-      case REAL -> putFloat(json, columnName, resultSet, colIndex);
+      case FLOAT, DOUBLE, REAL -> putDouble(json, columnName, resultSet, colIndex);
       case NUMERIC, DECIMAL -> putBigDecimal(json, columnName, resultSet, colIndex);
       case CHAR, NVARCHAR, VARCHAR, LONGVARCHAR -> putString(json, columnName, resultSet, colIndex);
       case DATE -> putDate(json, columnName, resultSet, colIndex);
@@ -90,6 +91,7 @@ private void putValue(final JDBCType columnType,
 
   @Override
   public JDBCType getDatabaseFieldType(final JsonNode field) {
+    // throw new RuntimeException("SGX");
     try {
       final String typeName = field.get(INTERNAL_COLUMN_TYPE_NAME).asText();
       if (typeName.equalsIgnoreCase("geography")
@@ -187,4 +189,14 @@ protected void setTimestampWithTimezone(final PreparedStatement preparedStatemen
     }
   }
 
+  protected void setReal(final PreparedStatement preparedStatement, final int parameterIndex, final String value) throws SQLException {
+    preparedStatement.setFloat(parameterIndex, Float.parseFloat(value));
+    throw new RuntimeException("SGX");
+  }
+
+  protected void putFloat(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException {
+    node.put(columnName, DataTypeUtils.returnNullIfInvalid(() -> resultSet.getFloat(index), Float::isFinite));
+    throw new RuntimeException("SGX");
+  }
+
 }