[Cherry-pick to branch-1.3] [#13973][#13974] fix(catalog-jdbc): Escape column default literals and table properties (#13975)

catalogs-contrib/catalog-jdbc-hologres/src/main/java/org/apache/gravitino/catalog/hologres/converter/HologresColumnDefaultValueConverter.java

@@ -46,7 +46,8 @@ public HologresColumnDefaultValueConverter() {}
   // for example:
   // '2021-10-01 00:00:00'::timestamp without time zone
   // ''something'::character varying'
-  private static final Pattern VALUE_WITH_TYPE = Pattern.compile("'(.+?)'::\\s*(.+)");
+  // Possessive repetition avoids recursive backtracking for long default literals.
+  private static final Pattern VALUE_WITH_TYPE = Pattern.compile("'((?:[^']|'')*+)'::\\s*(.+)");
 
   // for example:
   // NULL::character varying
@@ -86,10 +87,20 @@ public Expression toGravitino(
     }
   }
 
+  /**
+   * Doubles single quotes only. With {@code standard_conforming_strings} on, the default since
+   * PostgreSQL 9.1, a backslash in a string literal is an ordinary character.
+   */
+  @Override
+  protected String quoteStringLiteral(String value) {
+    return "'" + value.replace("'", "''") + "'";
+  }
+
   private Expression parseLiteral(JdbcTypeConverter.JdbcTypeBean type, String columnDefaultValue) {
     Matcher matcher = VALUE_WITH_TYPE.matcher(columnDefaultValue);
     if (matcher.find()) {
-      columnDefaultValue = matcher.group(1);
+      // String literals in the catalog are shown with embedded single quotes doubled.
+      columnDefaultValue = matcher.group(1).replace("''", "'");
     }
 
     if (columnDefaultValue.startsWith(NULL_PREFIX)) {

catalogs-contrib/catalog-jdbc-hologres/src/test/java/org/apache/gravitino/catalog/hologres/converter/TestHologresColumnDefaultValueConverter.java

@@ -34,6 +34,17 @@ public class TestHologresColumnDefaultValueConverter {
   private final HologresColumnDefaultValueConverter converter =
       new HologresColumnDefaultValueConverter();
 
+  /** Verifies long quoted text defaults can be loaded without recursive regex stack growth. */
+  @Test
+  public void testLongQuotedTextLiteralIsUnescaped() {
+    String value = "text '::quoted \\path ".repeat(2000);
+    JdbcTypeConverter.JdbcTypeBean type = new JdbcTypeConverter.JdbcTypeBean("text");
+    Assertions.assertEquals(
+        Literals.stringLiteral(value),
+        converter.toGravitino(
+            type, converter.fromGravitino(Literals.stringLiteral(value)) + "::text", false, true));
+  }
+
   @Test
   public void testNullDefaultValueNullable() {
     JdbcTypeConverter.JdbcTypeBean type = new JdbcTypeConverter.JdbcTypeBean("int4");
@@ -206,4 +217,20 @@ public void testFromGravitinoDefaultValueNotSet() {
     String result = converter.fromGravitino(DEFAULT_VALUE_NOT_SET);
     Assertions.assertNull(result);
   }
+
+  @Test
+  public void testStringLiteralDoublesQuotesOnly() {
+    Assertions.assertEquals(
+        "'O''Brien \\path'", converter.fromGravitino(Literals.stringLiteral("O'Brien \\path")));
+  }
+
+  @Test
+  public void testQuotedTextLiteralIsUnescaped() {
+    JdbcTypeConverter.JdbcTypeBean type = new JdbcTypeConverter.JdbcTypeBean("text");
+    Assertions.assertEquals(
+        Literals.stringLiteral("O'Brien'::x \\path"),
+        converter.toGravitino(type, "'O''Brien''::x \\path'::text", false, true));
+    Assertions.assertEquals(
+        Literals.stringLiteral(""), converter.toGravitino(type, "''::text", false, true));
+  }
 }

catalogs-contrib/catalog-jdbc-oceanbase/src/test/java/org/apache/gravitino/catalog/oceanbase/operation/TestOceanBaseTableOperations.java

@@ -1262,4 +1262,58 @@ public void testAlterTableSkipsEmptySql() {
     JdbcTable table = TABLE_OPERATIONS.load(TEST_DB_NAME, tableName);
     Assertions.assertEquals("Updated comment", table.columns()[0].comment());
   }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"O'Brien", "quote\\' and \\new\\path; \"double\"", "ends with \\"})
+  void testStringDefaultValueRoundTrips(String defaultValue) {
+    String tableName = RandomNameUtils.genRandomName("oceanbase_default_round_trip");
+    JdbcColumn column =
+        JdbcColumn.builder()
+            .withName("col1")
+            .withType(VARCHAR)
+            .withNullable(true)
+            .withDefaultValue(Literals.varcharLiteral(255, defaultValue))
+            .build();
+    TABLE_OPERATIONS.create(
+        TEST_DB_NAME,
+        tableName,
+        new JdbcColumn[] {column},
+        null,
+        Collections.emptyMap(),
+        Transforms.EMPTY_TRANSFORM,
+        Distributions.NONE,
+        Indexes.EMPTY_INDEXES);
+    try {
+      JdbcTable table = TABLE_OPERATIONS.load(TEST_DB_NAME, tableName);
+      Assertions.assertEquals(
+          Literals.varcharLiteral(255, defaultValue), table.columns()[0].defaultValue());
+
+      TABLE_OPERATIONS.alterTable(
+          TEST_DB_NAME,
+          tableName,
+          TableChange.addColumn(
+              new String[] {"col2"},
+              VARCHAR,
+              null,
+              TableChange.ColumnPosition.defaultPos(),
+              true,
+              false,
+              Literals.varcharLiteral(255, defaultValue)));
+      table = TABLE_OPERATIONS.load(TEST_DB_NAME, tableName);
+      Assertions.assertEquals(
+          Literals.varcharLiteral(255, defaultValue), table.columns()[1].defaultValue());
+
+      String updated = defaultValue + " updated";
+      TABLE_OPERATIONS.alterTable(
+          TEST_DB_NAME,
+          tableName,
+          TableChange.updateColumnDefaultValue(
+              new String[] {"col1"}, Literals.varcharLiteral(255, updated)));
+      table = TABLE_OPERATIONS.load(TEST_DB_NAME, tableName);
+      Assertions.assertEquals(
+          Literals.varcharLiteral(255, updated), table.columns()[0].defaultValue());
+    } finally {
+      TABLE_OPERATIONS.drop(TEST_DB_NAME, tableName);
+    }
+  }
 }

catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/converter/JdbcColumnDefaultValueConverter.java

@@ -18,8 +18,10 @@
  */
 package org.apache.gravitino.catalog.jdbc.converter;
 
+import static org.apache.gravitino.catalog.jdbc.utils.JdbcConnectorUtils.escapeSqlLiteral;
 import static org.apache.gravitino.rel.Column.DEFAULT_VALUE_NOT_SET;
 
+import java.math.BigDecimal;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.time.format.DateTimeFormatterBuilder;
@@ -77,18 +79,20 @@ public String fromGravitino(Expression defaultValue) {
         // case value is an empty string, we should wrap it with single quotes to avoid SQL error.
         if (StringUtils.isBlank(value)) {
           value = "'%s'".formatted(value);
+        } else {
+          value = validateNumericLiteral(value);
         }
         return value;
       } else if (type instanceof Types.TimestampType) {
         /**
          * @see LocalDateTime#toString() would return like 'yyyy-MM-ddTHH:mm:ss'
          */
         if (literal.value() instanceof String || literal.value() instanceof LocalDateTime) {
-          return String.format("'%s'", literal.value().toString().replace("T", " "));
+          return quoteStringLiteral(literal.value().toString().replace("T", " "));
         }
-        return String.format("'%s'", literal.value());
+        return quoteStringLiteral(String.valueOf(literal.value()));
       } else {
-        return String.format("'%s'", literal.value());
+        return quoteStringLiteral(String.valueOf(literal.value()));
       }
     }
 
@@ -102,4 +106,27 @@ public Expression toGravitino(
       boolean nullable) {
     return DEFAULT_VALUE_NOT_SET;
   }
+
+  /**
+   * Renders a value as a quoted SQL string literal for a column default. The default escapes
+   * backslashes and doubles single quotes, as MySQL-compatible dialects require. Dialects with
+   * different string literal rules override this method.
+   *
+   * @param value the unquoted literal value
+   * @return the quoted and escaped SQL string literal
+   */
+  protected String quoteStringLiteral(String value) {
+    return "'" + escapeSqlLiteral(value, '\'') + "'";
+  }
+
+  private static String validateNumericLiteral(String value) {
+    String trimmed = value.trim();
+    try {
+      new BigDecimal(trimmed);
+    } catch (NumberFormatException e) {
+      throw new IllegalArgumentException(
+          String.format("Invalid numeric column default value: %s", value), e);
+    }
+    return trimmed;
+  }
 }

catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/converter/TestJdbcColumnDefaultValueConverter.java

@@ -23,8 +23,11 @@
 import org.apache.gravitino.rel.Column;
 import org.apache.gravitino.rel.expressions.Expression;
 import org.apache.gravitino.rel.expressions.literals.Literals;
+import org.apache.gravitino.rel.types.Types;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 public class TestJdbcColumnDefaultValueConverter {
 
@@ -58,4 +61,46 @@ public void testFromGravitino() {
     result = converter.fromGravitino(expression);
     Assertions.assertEquals("1234", result);
   }
+
+  @Test
+  public void testStringLiteralIsEscaped() {
+    Assertions.assertEquals(
+        "'O''Brien'", converter.fromGravitino(Literals.stringLiteral("O'Brien")));
+    Assertions.assertEquals(
+        "'a\\\\'' , 1); DROP TABLE t; --'",
+        converter.fromGravitino(Literals.varcharLiteral(64, "a\\' , 1); DROP TABLE t; --")));
+    Assertions.assertEquals(
+        "'it''s'", converter.fromGravitino(Literals.of("it's", Types.FixedCharType.of(8))));
+    Assertions.assertEquals(
+        "'2024-01-01 00:00:00'''",
+        converter.fromGravitino(
+            Literals.of("2024-01-01T00:00:00'", Types.TimestampType.withoutTimeZone())));
+    Assertions.assertEquals(
+        "'2024-01-01'''",
+        converter.fromGravitino(Literals.of("2024-01-01'", Types.DateType.get())));
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"0", "-12", "+3", "1.50", " 42 ", "1.0E10", "-2.5e-3"})
+  public void testValidNumericLiteralIsWritten(String value) {
+    Assertions.assertEquals(
+        value.trim(), converter.fromGravitino(Literals.of(value, Types.DecimalType.of(20, 4))));
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"1); DROP TABLE t; --", "1 OR 1=1", "abc", "NaN", "0x10", "1,000"})
+  public void testInvalidNumericLiteralIsRejected(String value) {
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> converter.fromGravitino(Literals.of(value, Types.IntegerType.get())));
+    Assertions.assertEquals(
+        "Invalid numeric column default value: " + value, exception.getMessage());
+  }
+
+  @Test
+  public void testBlankNumericLiteralKeepsQuotedEmptyString() {
+    Assertions.assertEquals(
+        "''", converter.fromGravitino(Literals.of("", Types.IntegerType.get())));
+  }
 }

catalogs/catalog-jdbc-doris/src/main/java/org/apache/gravitino/catalog/doris/converter/DorisColumnDefaultValueConverter.java

@@ -32,6 +32,7 @@
 
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import javax.annotation.Nullable;
 import org.apache.gravitino.catalog.jdbc.converter.JdbcColumnDefaultValueConverter;
 import org.apache.gravitino.catalog.jdbc.converter.JdbcTypeConverter;
 import org.apache.gravitino.rel.expressions.Expression;
@@ -88,6 +89,84 @@ public String fromGravitinoForAddColumn(
     return super.fromGravitino(defaultValue);
   }
 
+  /**
+   * Converts a CREATE TABLE default value to Doris SQL.
+   *
+   * <p>Doris 2.1 and later keep the text of a string default unchanged when the table is created
+   * and parse it again as a double-quoted literal when a row takes the default, so string defaults
+   * must use that form there. Earlier versions decode the literal when the table is created and use
+   * the shared single-quoted form.
+   *
+   * @param defaultValue the Gravitino default value
+   * @param useDoubleQuotedLiteral whether string literals use double quotes with backslash escapes
+   * @return the Doris SQL representation, or {@code null} when the default is unset
+   */
+  @Nullable
+  public String fromGravitinoForCreateTable(
+      Expression defaultValue, boolean useDoubleQuotedLiteral) {
+    if (useDoubleQuotedLiteral && isNonNullStringLiteral(defaultValue)) {
+      return quoteDorisLiteral(String.valueOf(((Literal<?>) defaultValue).value()), false, false);
+    }
+    return super.fromGravitino(defaultValue);
+  }
+
+  /**
+   * Returns whether a default value is a non-null string, varchar or char literal.
+   *
+   * @param defaultValue the Gravitino default value
+   * @return whether the value is a non-null string literal
+   */
+  public static boolean isNonNullStringLiteral(Expression defaultValue) {
+    if (!(defaultValue instanceof Literal)) {
+      return false;
+    }
+    Literal<?> literal = (Literal<?>) defaultValue;
+    return literal.value() != null
+        && (literal.dataType() instanceof Types.StringType
+            || literal.dataType() instanceof Types.VarCharType
+            || literal.dataType() instanceof Types.FixedCharType);
+  }
+
+  /**
+   * Converts a loaded default value for a Doris MODIFY COLUMN definition.
+   *
+   * <p>String literals containing quote or backslash characters need Doris-specific escaping.
+   * Unparsed expressions come from native Doris metadata and are passed through unchanged.
+   *
+   * @param defaultValue the loaded Gravitino default value
+   * @param doubleEscapeBackslashes whether Doris 3.x requires an additional backslash-escaping
+   *     layer
+   * @param tripleEscapeQuotes whether Doris 3.x MODIFY COLUMN requires three backslashes before an
+   *     embedded double quote
+   * @return the Doris SQL representation, or {@code null} when the default is unset
+   */
+  @Nullable
+  public String fromGravitinoForModifyColumn(
+      Expression defaultValue, boolean doubleEscapeBackslashes, boolean tripleEscapeQuotes) {
+    if (DEFAULT_VALUE_NOT_SET.equals(defaultValue)) {
+      return null;
+    }
+
+    if (defaultValue instanceof UnparsedExpression) {
+      return ((UnparsedExpression) defaultValue).unparsedExpression();
+    }
+
+    if (defaultValue instanceof Literal) {
+      Literal<?> literal = (Literal<?>) defaultValue;
+      if (literal.value() != null
+          && (literal.dataType() instanceof Types.StringType
+              || literal.dataType() instanceof Types.VarCharType
+              || literal.dataType() instanceof Types.FixedCharType)) {
+        String value = String.valueOf(literal.value());
+        if (value.indexOf('\\') >= 0 || value.indexOf('\'') >= 0 || value.indexOf('"') >= 0) {
+          return quoteDorisModifyLiteral(value, doubleEscapeBackslashes, tripleEscapeQuotes);
+        }
+      }
+    }
+
+    return super.fromGravitino(defaultValue);
+  }
+
   @Override
   public Expression toGravitino(
       JdbcTypeConverter.JdbcTypeBean columnType,
@@ -162,6 +241,14 @@ private static String quoteDorisLiteral(
     return "\"" + escaped + "\"";
   }
 
+  private static String quoteDorisModifyLiteral(
+      String value, boolean doubleEscapeBackslashes, boolean tripleEscapeQuotes) {
+    String escapedBackslash = doubleEscapeBackslashes ? "\\".repeat(4) : "\\".repeat(2);
+    String escapedQuote = "\\".repeat(tripleEscapeQuotes ? 3 : 1) + "\"";
+    String escaped = value.replace("\\", escapedBackslash).replace("\"", escapedQuote);
+    return "\"" + escaped + "\"";
+  }
+
   private static String unescapeDorisLiteral(String value) {
     StringBuilder result = new StringBuilder(value.length());
     for (int i = 0; i < value.length(); i++) {

catalogs/catalog-jdbc-doris/src/main/java/org/apache/gravitino/catalog/doris/operation/DorisDatabaseOperations.java

@@ -28,6 +28,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.gravitino.catalog.doris.utils.DorisUtils;
 import org.apache.gravitino.catalog.jdbc.JdbcSchema;
@@ -40,6 +42,9 @@
 public class DorisDatabaseOperations extends JdbcDatabaseOperations {
   public static final String COMMENT_KEY = "comment";
 
+  private static final Pattern DATABASE_PROPERTY_PATTERN =
+      Pattern.compile("\"(.*?)\"\\s*=\\s*\"(.*)\"\\s*,?");
+
   @Override
   public String generateCreateDatabaseSql(
       String databaseName, String comment, Map<String, String> properties) {
@@ -130,7 +135,17 @@ protected Map<String, String> getDatabaseProperties(String databaseName) {
       throw new NoSuchTableException("Database %s does not exist.", databaseName);
     }
 
-    return DorisUtils.extractPropertiesFromSql(createDatabaseSql);
+    // SHOW CREATE DATABASE uses PrintableMap to wrap raw metadata in quotes without escaping it.
+    // Capture through the last quote on each property line and preserve the contents verbatim;
+    // SQL-literal unescaping would corrupt doubled quotes and backslashes.
+    Map<String, String> properties = new HashMap<>();
+    for (String line : createDatabaseSql.split("\n")) {
+      Matcher matcher = DATABASE_PROPERTY_PATTERN.matcher(line.trim());
+      if (matcher.matches()) {
+        properties.put(matcher.group(1), matcher.group(2));
+      }
+    }
+    return properties;
   }
 
   @Override