SNOW-918949: Add row index information to all exceptions

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.");
+          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.");
+          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);

src/main/java/net/snowflake/ingest/streaming/internal/DataValidationUtil.java

@@ -457,7 +457,10 @@ static TimestampWrapper validateAndParseTimestamp(
     if (offsetDateTime.getYear() < 1 || offsetDateTime.getYear() > 9999) {
       throw new SFException(
           ErrorCode.INVALID_VALUE_ROW,
-          "Timestamp out of representable inclusive range of years between 1 and 9999");
+          String.format(
+              "Timestamp out of representable inclusive range of years between 1 and 9999,"
+                  + " rowIndex:%d",
+              insertRowIndex));
     }
     return new TimestampWrapper(offsetDateTime, scale);
   }
@@ -588,7 +591,10 @@ static int validateAndParseDate(String columnName, Object input, long insertRowI
     if (offsetDateTime.getYear() < -9999 || offsetDateTime.getYear() > 9999) {
       throw new SFException(
           ErrorCode.INVALID_VALUE_ROW,
-          "Date out of representable inclusive range of years between -9999 and 9999");
+          String.format(
+              "Date out of representable inclusive range of years between -9999 and 9999,"
+                  + " rowIndex:%d",
+              insertRowIndex));
     }
 
     return Math.toIntExact(offsetDateTime.toLocalDate().toEpochDay());
@@ -814,7 +820,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:%s",
+              "Number out of representable exclusive range of (-1e%s..1e%s), rowIndex:%d",
               precision - scale, precision - scale, insertRowIndex));
     }
   }
@@ -858,8 +864,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:%s",
+            "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))));
@@ -882,7 +887,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: %s, reason:"
+            "Value cannot be ingested into Snowflake column %s of type %s, rowIndex:%d, reason:"
                 + " %s",
             columnName, snowflakeType, rowIndex, reason));
   }

src/main/java/net/snowflake/ingest/streaming/internal/ParquetRowBuffer.java

@@ -210,8 +210,8 @@ private float addRow(
       throw new SFException(
           ErrorCode.MAX_ROW_SIZE_EXCEEDED,
           String.format(
-              "rowSizeInBytes=%.3f maxAllowedRowSizeInBytes=%d",
-              size, clientBufferParameters.getMaxAllowedRowSizeInBytes()));
+              "rowSizeInBytes:%.3f, maxAllowedRowSizeInBytes:%d, rowIndex:%d",
+              size, clientBufferParameters.getMaxAllowedRowSizeInBytes(), insertRowsCurrIndex));
     }
 
     out.accept(Arrays.asList(indexedRow));