update message

snowflakedb · Sep 27, 2023 · 22fe148 · 22fe148
1 parent aea5519
commit 22fe148
Show file tree

Hide file tree

Showing 6 changed files with 52 additions and 54 deletions.
diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/AbstractRowBuffer.java b/src/main/java/net/snowflake/ingest/streaming/internal/AbstractRowBuffer.java
@@ -236,10 +236,11 @@ public float getSize() {
    *
    * @param row the input row
    * @param error the insert error that we return to the customer
+   * @param rowIndex the index of the current row in the input batch
    * @return the set of input column names
    */
   Set<String> verifyInputColumns(
-      Map<String, Object> row, InsertValidationResponse.InsertError error) {
+      Map<String, Object> row, InsertValidationResponse.InsertError error, int rowIndex) {
     // Map of unquoted column name -> original column name
     Map<String, String> inputColNamesMap =
         row.keySet().stream()
@@ -260,7 +261,8 @@ Set<String> verifyInputColumns(
       throw new SFException(
           ErrorCode.INVALID_FORMAT_ROW,
           "Extra columns: " + extraCols,
-          "Columns not present in the table shouldn't be specified. Row Index:%d");
+          String.format(
+              "Columns not present in the table shouldn't be specified, rowIndex:%d", rowIndex));
     }
 
     // Check for missing columns in the row
@@ -278,7 +280,8 @@ Set<String> verifyInputColumns(
       throw new SFException(
           ErrorCode.INVALID_FORMAT_ROW,
           "Missing columns: " + missingCols,
-          "Values for all non-nullable columns must be specified. Row Index:%d");
+          String.format(
+              "Values for all non-nullable columns must be specified, rowIndex:%d", rowIndex));
     }
 
     return inputColNamesMap.keySet();
@@ -304,12 +307,12 @@ public InsertValidationResponse insertRows(
       this.channelState.updateInsertStats(System.currentTimeMillis(), this.bufferedRowCount);
       if (onErrorOption == OpenChannelRequest.OnErrorOption.CONTINUE) {
         // Used to map incoming row(nth row) to InsertError(for nth row) in response
-        long rowIndex = 0;
+        int rowIndex = 0;
         for (Map<String, Object> row : rows) {
           InsertValidationResponse.InsertError error =
               new InsertValidationResponse.InsertError(row, rowIndex);
           try {
-            Set<String> inputColumnNames = verifyInputColumns(row, error);
+            Set<String> inputColumnNames = verifyInputColumns(row, error, rowIndex);
             rowsSizeInBytes +=
                 addRow(row, this.bufferedRowCount, this.statsMap, inputColumnNames, rowIndex);
             this.bufferedRowCount++;
@@ -333,7 +336,7 @@ public InsertValidationResponse insertRows(
         float tempRowsSizeInBytes = 0F;
         int tempRowCount = 0;
         for (Map<String, Object> row : rows) {
-          Set<String> inputColumnNames = verifyInputColumns(row, null);
+          Set<String> inputColumnNames = verifyInputColumns(row, null, tempRowCount);
           tempRowsSizeInBytes +=
               addTempRow(row, tempRowCount, this.tempStatsMap, inputColumnNames, tempRowCount);
           checkBatchSizeEnforcedMaximum(tempRowsSizeInBytes);

diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/DataValidationUtil.java b/src/main/java/net/snowflake/ingest/streaming/internal/DataValidationUtil.java
@@ -24,11 +24,7 @@
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeParseException;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
 import java.util.function.Supplier;
 import net.snowflake.client.jdbc.internal.google.common.collect.Sets;
 import net.snowflake.client.jdbc.internal.snowflake.common.core.SnowflakeDateTimeFormat;
@@ -458,8 +454,8 @@ static TimestampWrapper validateAndParseTimestamp(
       throw new SFException(
           ErrorCode.INVALID_VALUE_ROW,
           String.format(
-              "Timestamp out of representable inclusive range of years between 1 and 9999, Row"
-                  + " Index:%d",
+              "Timestamp out of representable inclusive range of years between 1 and 9999,"
+                  + " rowIndex:%d",
               insertRowIndex));
     }
     return new TimestampWrapper(offsetDateTime, scale);
@@ -592,8 +588,8 @@ static int validateAndParseDate(String columnName, Object input, long insertRowI
       throw new SFException(
           ErrorCode.INVALID_VALUE_ROW,
           String.format(
-              "Date out of representable inclusive range of years between -9999 and 9999, Row"
-                  + " Index:%d",
+              "Date out of representable inclusive range of years between -9999 and 9999,"
+                  + " rowIndex:%d",
               insertRowIndex));
     }
 
@@ -820,7 +816,7 @@ static void checkValueInRange(
       throw new SFException(
           ErrorCode.INVALID_FORMAT_ROW,
           String.format(
-              "Number out of representable exclusive range of (-1e%s..1e%s), Row Index:%d",
+              "Number out of representable exclusive range of (-1e%s..1e%s), rowIndex:%d",
               precision - scale, precision - scale, insertRowIndex));
     }
   }
@@ -864,8 +860,7 @@ private static SFException typeNotAllowedException(
     return new SFException(
         ErrorCode.INVALID_FORMAT_ROW,
         String.format(
-            "Object of type %s cannot be ingested into Snowflake column %s of type %s, Row"
-                + " Index:%d",
+            "Object of type %s cannot be ingested into Snowflake column %s of type %s, rowIndex:%d",
             javaType.getName(), columnName, snowflakeType, insertRowIndex),
         String.format(
             String.format("Allowed Java types: %s", String.join(", ", allowedJavaTypes))));
@@ -888,7 +883,7 @@ private static SFException valueFormatNotAllowedException(
     return new SFException(
         ErrorCode.INVALID_VALUE_ROW,
         String.format(
-            "Value cannot be ingested into Snowflake column %s of type %s, Row Index:%d, reason:"
+            "Value cannot be ingested into Snowflake column %s of type %s, rowIndex:%d, reason:"
                 + " %s",
             columnName, snowflakeType, rowIndex, reason));
   }

diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/ParquetRowBuffer.java b/src/main/java/net/snowflake/ingest/streaming/internal/ParquetRowBuffer.java
@@ -210,7 +210,7 @@ private float addRow(
       throw new SFException(
           ErrorCode.MAX_ROW_SIZE_EXCEEDED,
           String.format(
-              "rowSizeInBytes=%.3f, maxAllowedRowSizeInBytes=%d, Row Index=%d",
+              "rowSizeInBytes:%.3f, maxAllowedRowSizeInBytes:%d, rowIndex:%d",
               size, clientBufferParameters.getMaxAllowedRowSizeInBytes(), insertRowsCurrIndex));
     }