fix(ADH-5681): Explicitly specify SQL types for db, table, and partit…

…ion parameters in removeTxnComponents for compaction. fix(ADH-5681): Sanitize error message in compaction queue update. fix(ADH-5681): generate utf8Sanitize utility method and use it in CompactionTxnHandler.
arenadata · Feb 19, 2025 · 5225fe7 · 5225fe7
1 parent 3145987
commit 5225fe7
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 5 deletions.
diff --git a/...store-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java b/...store-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
@@ -453,6 +453,7 @@ public void markRefused(CompactionInfo info) throws MetaException {
   @Override
   @RetrySemantics.CannotRetry
   public void setCleanerRetryRetentionTimeOnError(CompactionInfo info) throws MetaException {
+    String sanitizedErrorMessage = TxnUtils.utf8Sanitize(info.errorMessage);
     if (info.isAbortedTxnCleanup() && info.id == 0) {
       /*
        * MUTEX_KEY.CompactionScheduler lock ensures that there is only 1 entry in
@@ -473,7 +474,7 @@ public void setCleanerRetryRetentionTimeOnError(CompactionInfo info) throws Meta
                 .addValue("type", Character.toString(thriftCompactionType2DbType(info.type)))
                 .addValue("state", Character.toString(info.state))
                 .addValue("retention", info.retryRetention)
-                .addValue("msg", info.errorMessage), null);
+                .addValue("msg", sanitizedErrorMessage), null);
         if (updCnt == 0) {
           LOG.error("Unable to update/insert compaction queue record: {}. updCnt={}", info, updCnt);
           throw new MetaException("Unable to insert abort retry entry into COMPACTION QUEUE: " +
@@ -487,7 +488,7 @@ public void setCleanerRetryRetentionTimeOnError(CompactionInfo info) throws Meta
           "UPDATE \"COMPACTION_QUEUE\" SET \"CQ_RETRY_RETENTION\" = :retention, \"CQ_ERROR_MESSAGE\"= :msg WHERE \"CQ_ID\" = :id",
           new MapSqlParameterSource()
               .addValue("retention", info.retryRetention)
-              .addValue("msg", info.errorMessage)
+              .addValue("msg", sanitizedErrorMessage)
               .addValue("id", info.id),
           ParameterizedCommand.EXACTLY_ONE_ROW);
     }

diff --git a/...tastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnUtils.java b/...tastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnUtils.java
@@ -696,4 +696,8 @@ public static String createInsertPreparedStmt(String tableName, List<String> col
     sb.append(") values (" + placeholder + ")");
     return sb.toString();
   }
+
+  public static String utf8Sanitize(String input) {
+    return input == null ? null : input.replace("\0", "");
+  }
 }
diff --git a/...rc/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/functions/MarkCleanedFunction.java b/...rc/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/functions/MarkCleanedFunction.java
@@ -122,9 +122,9 @@ private void removeTxnComponents(CompactionInfo info, MultiDataSourceJdbcResourc
 
     MapSqlParameterSource params = new MapSqlParameterSource()
         .addValue("state", TxnStatus.ABORTED.getSqlConst(), Types.CHAR)
-        .addValue("db", info.dbname)
-        .addValue("table", info.tableName)
-        .addValue("partition", info.partName);
+        .addValue("db", info.dbname, Types.VARCHAR)
+        .addValue("table", info.tableName, Types.VARCHAR)
+        .addValue("partition", info.partName, Types.VARCHAR);
 
     int totalCount = 0;
     if (!info.hasUncompactedAborts && info.highestWriteId != 0) {