This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 9b89234694 [Cherry-pick to branch-1.3] [#14608] fix(spark): convert
default SQL literals to typed values (#14609) (#14696)
9b89234694 is described below
commit 9b89234694e5f4a7f963c4bc3f8f4533e86192e1
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Nov 6 14:04:48 2026 +0800
[Cherry-pick to branch-1.3] [#14608] fix(spark): convert default SQL
literals to typed values (#14609) (#14696)
### What changes were proposed in this pull request?
Backport #14609 (`a93aa73eadef659a9550be276e721d461ffa6233`) to
`branch-1.3` and resolve the committed cherry-pick conflicts.
Keep typed default conversion in `SparkTableChangeConverter34`, which is
used by Spark 3.4 and 3.5. The shared converter retains Spark 3.3
compatibility through a lazy-schema overload. Move the default-value
regressions to the version-specific tests and add their required test
dependencies. Preserve the upstream JDBC binary encoding, timestamp
rendering, schema-loading and batch-validation behavior.
### Why are the changes needed?
Spark 3.3 does not provide the column-default update API used by the
main-branch fix. Copying it into the shared module prevents the 1.3
connector from compiling. Conflicted import/documentation context also
included main-only changes that do not belong in this backport.
### Does this PR introduce _any_ user-facing change?
Spark 3.4/3.5 default SQL literals are converted to typed values instead
of being stored as SQL text. Binary values survive REST serialization;
timestamp defaults follow the target column type and Spark session time
zone. Nonliteral expressions, unsupported DROP DEFAULT,
string-to-timestamp defaults and schema-changing batches combined with
schema-dependent defaults are rejected explicitly. Timestamp/date/string
defaults require a table-schema read, shared once per batch.
Spark 3.3 retains its existing supported table changes. The binary
encoding requires the updated server-side JDBC converters, and MySQL
zoned timestamp defaults require MySQL 8.0.19 or later, as documented in
#14609.
### How was this patch tested?
- All 141 targeted cases passed with zero failures, errors or skips: 11
shared Spark 3.3 cases, 36 Spark 3.4.3 cases, the same 36 default/schema
cases on the Spark 3.5.3 runtime, 10 DTO cases, 27 JDBC converter cases,
9 live MySQL timestamp cases and 12 PostgreSQL cases (including 3 live
binary-default cases).
- The Spark 3.5 run reused the 3.4 regression classes through a local
Gradle init script and verified the resolved Catalyst runtime was 3.5.3.
- Relevant Spotless checks, Spark 3.3 module compilation,
conflict-marker scan and `git diff --check` passed.
Remote CI for the conflict-resolution commit is pending. No deployed
Spark-to-server/database end-to-end run or full core suite was
performed.
---------
Co-authored-by: Qi Yu <[email protected]>
---
.../converter/JdbcColumnDefaultValueConverter.java | 46 ++-
.../TestJdbcColumnDefaultValueConverter.java | 74 ++++
catalogs/catalog-jdbc-mysql/build.gradle.kts | 1 +
.../mysql/operation/TestMysqlTableOperations.java | 98 ++++++
.../PostgreSqlColumnDefaultValueConverter.java | 8 +
.../TestPostgreSqlColumnDefaultValueConverter.java | 16 +
.../operation/TestPostgreSqlTableOperations.java | 49 +++
.../apache/gravitino/dto/util/DTOConverters.java | 7 +
.../gravitino/dto/util/TestDTOConverters.java | 19 +
docs/expression.md | 32 ++
.../spark/connector/SparkTableChangeConverter.java | 17 +
.../spark/connector/catalog/BaseCatalog.java | 31 +-
.../catalog/TestTransformTableChange.java | 14 +
spark-connector/v3.4/spark/build.gradle.kts | 1 +
.../connector/SparkTableChangeConverter34.java | 143 +++++++-
.../connector/TestSparkTableChangeConverter34.java | 385 ++++++++++++++++++++-
.../catalog/TestBaseCatalogDefaults34.java | 284 +++++++++++++++
17 files changed, 1195 insertions(+), 30 deletions(-)
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/converter/JdbcColumnDefaultValueConverter.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/converter/JdbcColumnDefaultValueConverter.java
index ac3099a401..f4eecc197b 100644
---
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/converter/JdbcColumnDefaultValueConverter.java
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/converter/JdbcColumnDefaultValueConverter.java
@@ -21,11 +21,12 @@ 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 com.google.common.io.BaseEncoding;
import java.math.BigDecimal;
-import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.time.temporal.ChronoField;
+import java.util.Locale;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.rel.expressions.Expression;
import org.apache.gravitino.rel.expressions.FunctionExpression;
@@ -83,14 +84,35 @@ public class JdbcColumnDefaultValueConverter {
value = validateNumericLiteral(value);
}
return value;
+ } else if (type instanceof Types.BinaryType && literal.value()
instanceof byte[]) {
+ return binaryLiteral((byte[]) literal.value());
+ } else if (type instanceof Types.BinaryType
+ && literal.value() instanceof String
+ && ((String) literal.value()).startsWith("0x")) {
+ byte[] value;
+ try {
+ value =
+ BaseEncoding.base16()
+ .lowerCase()
+ .decode(((String)
literal.value()).substring(2).toLowerCase(Locale.ROOT));
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ "Invalid binary column default value: use 0x followed by
hexadecimal byte pairs", e);
+ }
+ return binaryLiteral(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 quoteStringLiteral(literal.value().toString().replace("T", "
"));
+ // java.time omits seconds for whole-minute values, but MySQL requires
them.
+ String value =
+ literal
+ .value()
+ .toString()
+ .replace("T", " ")
+ .replaceFirst("^(\\d{4}-\\d{2}-\\d{2}
\\d{2}:\\d{2})(?=$|Z|[+-])", "$1:00");
+ if (((Types.TimestampType) type).hasTimeZone() && value.endsWith("Z"))
{
+ // MySQL accepts numeric offsets but rejects the equivalent ISO UTC
suffix.
+ value = value.substring(0, value.length() - 1) + "+00:00";
}
- return quoteStringLiteral(String.valueOf(literal.value()));
+ return quoteStringLiteral(value);
} else {
return quoteStringLiteral(String.valueOf(literal.value()));
}
@@ -119,6 +141,16 @@ public class JdbcColumnDefaultValueConverter {
return "'" + escapeSqlLiteral(value, '\'') + "'";
}
+ /**
+ * Renders binary bytes as a SQL literal. Dialects can override their binary
literal syntax.
+ *
+ * @param value the bytes to preserve
+ * @return a SQL expression containing the binary value
+ */
+ protected String binaryLiteral(byte[] value) {
+ return "X'" + BaseEncoding.base16().lowerCase().encode(value) + "'";
+ }
+
private static String validateNumericLiteral(String value) {
String trimmed = value.trim();
try {
diff --git
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/converter/TestJdbcColumnDefaultValueConverter.java
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/converter/TestJdbcColumnDefaultValueConverter.java
index 14070fafe2..a292ff0e6c 100644
---
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/converter/TestJdbcColumnDefaultValueConverter.java
+++
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/converter/TestJdbcColumnDefaultValueConverter.java
@@ -20,6 +20,9 @@ package org.apache.gravitino.catalog.jdbc.converter;
import static org.apache.gravitino.rel.Column.DEFAULT_VALUE_NOT_SET;
+import com.google.common.io.BaseEncoding;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
import org.apache.gravitino.rel.Column;
import org.apache.gravitino.rel.expressions.Expression;
import org.apache.gravitino.rel.expressions.literals.Literals;
@@ -33,6 +36,77 @@ public class TestJdbcColumnDefaultValueConverter {
private final JdbcColumnDefaultValueConverter converter = new
JdbcColumnDefaultValueConverter();
+ @ParameterizedTest
+ @ValueSource(strings = {"Z", "+08:00", "-07:00"})
+ void testTimestampDefaultsUseNumericOffsets(String offset) {
+ String value = "2026-11-04T12:34:56.123456" + offset;
+ String expected =
+ "'2026-11-04 12:34:56.123456" + (offset.equals("Z") ? "+00:00" :
offset) + "'";
+ Assertions.assertEquals(
+ expected,
+ converter.fromGravitino(
+ Literals.of(OffsetDateTime.parse(value),
Types.TimestampType.withTimeZone())));
+ Assertions.assertEquals(
+ expected, converter.fromGravitino(Literals.of(value,
Types.TimestampType.withTimeZone())));
+ Assertions.assertEquals(
+ "'2026-11-04 12:34:56.123456'",
+
converter.fromGravitino(Literals.timestampLiteral("2026-11-04T12:34:56.123456")));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"12:34:00", "00:00:00"})
+ void testWholeMinuteTimestampDefaultsIncludeSeconds(String time) {
+ String text = "2026-11-04T" + time;
+ for (String offset : new String[] {"Z", "+08:00", "-07:00"}) {
+ String expected = "'2026-11-04 " + time + (offset.equals("Z") ? "+00:00"
: offset) + "'";
+ OffsetDateTime value = OffsetDateTime.parse(text + offset);
+ Assertions.assertEquals(
+ expected,
+ converter.fromGravitino(Literals.of(value,
Types.TimestampType.withTimeZone())));
+ Assertions.assertEquals(
+ expected,
+ converter.fromGravitino(
+ Literals.of(value.toString(),
Types.TimestampType.withTimeZone())));
+ }
+ Assertions.assertEquals(
+ "'2026-11-04 " + time + "'",
+
converter.fromGravitino(Literals.timestampLiteral(LocalDateTime.parse(text))));
+ Assertions.assertEquals(
+ "'2026-11-04 " + time + "'",
+ converter.fromGravitino(
+ Literals.of(
+ LocalDateTime.parse(text).toString(),
Types.TimestampType.withoutTimeZone())));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"4142", "0080ff", ""})
+ void testBinaryDefaultsPreserveBytes(String hex) {
+ Assertions.assertEquals(
+ "X'" + hex + "'",
+ converter.fromGravitino(
+ Literals.of(BaseEncoding.base16().lowerCase().decode(hex),
Types.BinaryType.get())));
+ Assertions.assertEquals(
+ "X'" + hex + "'", converter.fromGravitino(Literals.of("0x" + hex,
Types.BinaryType.get())));
+ Assertions.assertEquals(
+ "'legacy'", converter.fromGravitino(Literals.of("legacy",
Types.BinaryType.get())));
+ Assertions.assertEquals("'0x4142'",
converter.fromGravitino(Literals.stringLiteral("0x4142")));
+ Assertions.assertEquals(
+ "X'00ff'", converter.fromGravitino(Literals.of("0x00FF",
Types.BinaryType.get())));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"0x0", "0xzz", "0x41'); DROP TABLE t; --"})
+ void testMalformedBinaryDefaultsAreRejected(String value) {
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> converter.fromGravitino(Literals.of(value,
Types.BinaryType.get())));
+ Assertions.assertEquals(
+ "Invalid binary column default value: use 0x followed by hexadecimal
byte pairs",
+ exception.getMessage());
+ Assertions.assertNotNull(exception.getCause());
+ }
+
@Test
public void testFromGravitino() {
Expression expression = DEFAULT_VALUE_NOT_SET;
diff --git a/catalogs/catalog-jdbc-mysql/build.gradle.kts
b/catalogs/catalog-jdbc-mysql/build.gradle.kts
index 712e0b5282..1671975993 100644
--- a/catalogs/catalog-jdbc-mysql/build.gradle.kts
+++ b/catalogs/catalog-jdbc-mysql/build.gradle.kts
@@ -51,6 +51,7 @@ dependencies {
testImplementation(project(":server-common"))
testImplementation(libs.awaitility)
+ testImplementation(libs.commons.dbcp2)
testImplementation(libs.junit.jupiter.api)
testImplementation(libs.junit.jupiter.params)
testImplementation(libs.mysql.driver)
diff --git
a/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperations.java
b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperations.java
index 68d37eb09c..0f0ff3b6e0 100644
---
a/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperations.java
+++
b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperations.java
@@ -21,9 +21,13 @@ package org.apache.gravitino.catalog.mysql.operation;
import static
org.apache.gravitino.catalog.mysql.MysqlTablePropertiesMetadata.MYSQL_AUTO_INCREMENT_OFFSET_KEY;
import static
org.apache.gravitino.catalog.mysql.MysqlTablePropertiesMetadata.MYSQL_ENGINE_KEY;
+import java.math.BigDecimal;
import java.sql.Connection;
+import java.sql.ResultSet;
import java.sql.Statement;
import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -32,11 +36,15 @@ import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import javax.sql.DataSource;
+import org.apache.commons.dbcp2.BasicDataSource;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.catalog.jdbc.JdbcColumn;
import org.apache.gravitino.catalog.jdbc.JdbcTable;
import org.apache.gravitino.catalog.jdbc.utils.DataSourceUtils;
+import
org.apache.gravitino.catalog.mysql.converter.MysqlColumnDefaultValueConverter;
+import org.apache.gravitino.catalog.mysql.converter.MysqlTypeConverter;
+import org.apache.gravitino.dto.util.DTOConverters;
import org.apache.gravitino.rel.Column;
import org.apache.gravitino.rel.TableChange;
import org.apache.gravitino.rel.expressions.distributions.Distributions;
@@ -52,6 +60,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
@Tag("gravitino-docker-test")
@@ -59,6 +68,95 @@ public class TestMysqlTableOperations extends TestMysql {
private static final Type VARCHAR = Types.VarCharType.of(255);
private static final Type INT = Types.IntegerType.get();
+ @ParameterizedTest
+ @CsvSource({
+ "+00:00, 12:34:56.123456", "+08:00, 12:34:56.123456", "-07:00,
12:34:56.123456",
+ "+00:00, 12:34:00", "+08:00, 12:34:00", "-07:00, 12:34:00",
+ "+00:00, 00:00:00", "+08:00, 00:00:00", "-07:00, 00:00:00"
+ })
+ void testZonedTimestampDefaultsPreserveInstantOnCreateAndAlter(String
timeZone, String time)
+ throws Exception {
+ for (boolean zoned : new boolean[] {true, false}) {
+ String table = RandomNameUtils.genRandomName("timestamp_defaults");
+ OffsetDateTime value = OffsetDateTime.parse("2026-11-04T" + time + "Z");
+ try (BasicDataSource dataSource =
+ (BasicDataSource)
DataSourceUtils.createDataSource(getMySQLCatalogProperties())) {
+ dataSource.setConnectionInitSqls(List.of("SET time_zone = '" +
timeZone + "'"));
+ MysqlTableOperations operations = new MysqlTableOperations();
+ operations.initialize(
+ dataSource,
+ JDBC_EXCEPTION_CONVERTER,
+ new MysqlTypeConverter(),
+ new MysqlColumnDefaultValueConverter(),
+ Collections.emptyMap());
+ operations.create(
+ TEST_DB_NAME.toString(),
+ table,
+ new JdbcColumn[] {
+ JdbcColumn.builder()
+ .withName("ts")
+ .withType(
+ zoned
+ ? Types.TimestampType.withTimeZone(6)
+ : Types.TimestampType.withoutTimeZone(6))
+ .withNullable(true)
+ .withDefaultValue(
+ zoned
+ ? Literals.of(value,
Types.TimestampType.withTimeZone())
+ : Literals.timestampLiteral(value.toLocalDateTime()))
+ .build()
+ },
+ null,
+ Collections.emptyMap(),
+ null,
+ Distributions.NONE,
+ Indexes.EMPTY_INDEXES);
+ try (Connection connection = dataSource.getConnection();
+ Statement statement = connection.createStatement()) {
+ String qualified = "`" + TEST_DB_NAME + "`.`" + table + "`";
+ for (int iteration = 0; iteration < 2; iteration++) {
+ if (iteration == 1) {
+ statement.executeUpdate("DELETE FROM " + qualified);
+ value = value.plusDays(1);
+ operations.alterTable(
+ TEST_DB_NAME.toString(),
+ table,
+ TableChange.updateColumnDefaultValue(
+ new String[] {"ts"},
+ DTOConverters.toFunctionArg(
+ zoned
+ ? Literals.of(value,
Types.TimestampType.withTimeZone())
+ :
Literals.timestampLiteral(value.toLocalDateTime()))));
+ }
+ statement.executeUpdate("INSERT INTO " + qualified + " () VALUES
()");
+ try (ResultSet result =
+ statement.executeQuery(
+ "SELECT UNIX_TIMESTAMP(ts), @@session.time_zone,
DATE_FORMAT(ts, '%Y-%m-%d %H:%i:%s.%f') FROM "
+ + qualified)) {
+ Assertions.assertTrue(result.next());
+ if (zoned) {
+ Assertions.assertEquals(
+ 0,
+ BigDecimal.valueOf(value.toEpochSecond())
+ .add(BigDecimal.valueOf(value.getNano(), 9))
+ .compareTo(result.getBigDecimal(1)));
+ } else {
+ Assertions.assertEquals(
+ value
+ .toLocalDateTime()
+ .format(DateTimeFormatter.ofPattern("yyyy-MM-dd
HH:mm:ss.SSSSSS")),
+ result.getString(3));
+ }
+ Assertions.assertEquals(timeZone, result.getString(2));
+ }
+ }
+ } finally {
+ operations.drop(TEST_DB_NAME.toString(), table);
+ }
+ }
+ }
+ }
+
@Test
public void testOperationTable() {
String tableName = RandomStringUtils.randomAlphabetic(16) + "_op_table";
diff --git
a/catalogs/catalog-jdbc-postgresql/src/main/java/org/apache/gravitino/catalog/postgresql/converter/PostgreSqlColumnDefaultValueConverter.java
b/catalogs/catalog-jdbc-postgresql/src/main/java/org/apache/gravitino/catalog/postgresql/converter/PostgreSqlColumnDefaultValueConverter.java
index 7dc3dc5398..e475489895 100644
---
a/catalogs/catalog-jdbc-postgresql/src/main/java/org/apache/gravitino/catalog/postgresql/converter/PostgreSqlColumnDefaultValueConverter.java
+++
b/catalogs/catalog-jdbc-postgresql/src/main/java/org/apache/gravitino/catalog/postgresql/converter/PostgreSqlColumnDefaultValueConverter.java
@@ -22,6 +22,7 @@ import static
org.apache.gravitino.catalog.jdbc.converter.JdbcTypeConverter.VARC
import static org.apache.gravitino.rel.Column.DEFAULT_VALUE_NOT_SET;
import static
org.apache.gravitino.rel.Column.DEFAULT_VALUE_OF_CURRENT_TIMESTAMP;
+import com.google.common.io.BaseEncoding;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
@@ -94,6 +95,13 @@ public class PostgreSqlColumnDefaultValueConverter extends
JdbcColumnDefaultValu
return "'" + value.replace("'", "''") + "'";
}
+ /** {@inheritDoc} */
+ @Override
+ protected String binaryLiteral(byte[] value) {
+ // PostgreSQL's X'...' literal is a bit string, so decode hex explicitly
for bytea.
+ return "decode('" + BaseEncoding.base16().lowerCase().encode(value) + "',
'hex')";
+ }
+
private Expression parseLiteral(JdbcTypeConverter.JdbcTypeBean type, String
columnDefaultValue) {
Matcher matcher = VALUE_WITH_TYPE.matcher(columnDefaultValue);
if (matcher.find()) {
diff --git
a/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/converter/TestPostgreSqlColumnDefaultValueConverter.java
b/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/converter/TestPostgreSqlColumnDefaultValueConverter.java
index dcf29eab48..f23c6ad07d 100644
---
a/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/converter/TestPostgreSqlColumnDefaultValueConverter.java
+++
b/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/converter/TestPostgreSqlColumnDefaultValueConverter.java
@@ -18,16 +18,32 @@
*/
package org.apache.gravitino.catalog.postgresql.converter;
+import com.google.common.io.BaseEncoding;
import org.apache.gravitino.catalog.jdbc.converter.JdbcTypeConverter;
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 TestPostgreSqlColumnDefaultValueConverter {
private final PostgreSqlColumnDefaultValueConverter converter =
new PostgreSqlColumnDefaultValueConverter();
+ @ParameterizedTest
+ @ValueSource(strings = {"4142", "0080ff", ""})
+ void testBinaryDefaultsUseByteaSyntax(String hex) {
+ Assertions.assertEquals(
+ "decode('" + hex + "', 'hex')",
+ converter.fromGravitino(
+ Literals.of(BaseEncoding.base16().lowerCase().decode(hex),
Types.BinaryType.get())));
+ Assertions.assertEquals(
+ "decode('" + hex + "', 'hex')",
+ converter.fromGravitino(Literals.of("0x" + hex,
Types.BinaryType.get())));
+ }
+
@Test
void testNulStringDefaultIsRejectedBeforeDdl() {
IllegalArgumentException error =
diff --git
a/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/operation/TestPostgreSqlTableOperations.java
b/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/operation/TestPostgreSqlTableOperations.java
index 74374f9193..58c9ec2eff 100644
---
a/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/operation/TestPostgreSqlTableOperations.java
+++
b/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/operation/TestPostgreSqlTableOperations.java
@@ -19,6 +19,7 @@
package org.apache.gravitino.catalog.postgresql.operation;
import com.google.common.collect.Maps;
+import com.google.common.io.BaseEncoding;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -40,6 +41,7 @@ import
org.apache.gravitino.catalog.jdbc.utils.DataSourceUtils;
import org.apache.gravitino.catalog.jdbc.utils.JdbcConnectorUtils;
import
org.apache.gravitino.catalog.postgresql.converter.PostgreSqlColumnDefaultValueConverter;
import
org.apache.gravitino.catalog.postgresql.converter.PostgreSqlTypeConverter;
+import org.apache.gravitino.dto.util.DTOConverters;
import org.apache.gravitino.exceptions.ForbiddenException;
import org.apache.gravitino.exceptions.GravitinoRuntimeException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
@@ -69,6 +71,53 @@ public class TestPostgreSqlTableOperations extends
TestPostgreSql {
private static Type VARCHAR = Types.VarCharType.of(255);
private static Type INT = Types.IntegerType.get();
+ @ParameterizedTest
+ @ValueSource(strings = {"4142", "0080ff", ""})
+ void testBinaryDefaultsPreserveBytesOnCreateAndAlter(String hex) throws
SQLException {
+ String table = RandomNameUtils.genRandomName("binary_defaults");
+ byte[] bytes = BaseEncoding.base16().lowerCase().decode(hex);
+ TABLE_OPERATIONS.create(
+ TEST_DB_NAME,
+ table,
+ new JdbcColumn[] {
+ JdbcColumn.builder()
+ .withName("payload")
+ .withType(Types.BinaryType.get())
+ .withNullable(true)
+ .withDefaultValue(Literals.of(bytes, Types.BinaryType.get()))
+ .build()
+ },
+ null,
+ Collections.emptyMap(),
+ null,
+ Distributions.NONE,
+ Indexes.EMPTY_INDEXES);
+ try (Connection connection = DATA_SOURCE.getConnection();
+ Statement statement = connection.createStatement()) {
+ String qualified = TEST_DB_NAME + "." + table;
+ statement.executeUpdate("INSERT INTO " + qualified + " DEFAULT VALUES");
+ try (ResultSet result = statement.executeQuery("SELECT payload FROM " +
qualified)) {
+ Assertions.assertTrue(result.next());
+ Assertions.assertArrayEquals(bytes, result.getBytes(1));
+ }
+ statement.executeUpdate("DELETE FROM " + qualified);
+ byte[] altered = BaseEncoding.base16().lowerCase().decode(hex + "ff");
+ TABLE_OPERATIONS.alterTable(
+ TEST_DB_NAME,
+ table,
+ TableChange.updateColumnDefaultValue(
+ new String[] {"payload"},
+ DTOConverters.toFunctionArg(Literals.of(altered,
Types.BinaryType.get()))));
+ statement.executeUpdate("INSERT INTO " + qualified + " DEFAULT VALUES");
+ try (ResultSet result = statement.executeQuery("SELECT payload FROM " +
qualified)) {
+ Assertions.assertTrue(result.next());
+ Assertions.assertArrayEquals(altered, result.getBytes(1));
+ }
+ } finally {
+ TABLE_OPERATIONS.drop(TEST_DB_NAME, table);
+ }
+ }
+
@ParameterizedTest
@ValueSource(strings = {"quoted\";CREATE TABLE injected_marker (id int);--"})
void testQuotedTableNamesStayIdentifiersDuringAlter(String name) {
diff --git
a/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
b/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
index e9ee2226dd..eed5e2c57a 100644
--- a/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
+++ b/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
@@ -20,6 +20,7 @@ package org.apache.gravitino.dto.util;
import static
org.apache.gravitino.rel.expressions.transforms.Transforms.NAME_OF_IDENTITY;
+import com.google.common.io.BaseEncoding;
import java.util.Arrays;
import java.util.Map;
import java.util.Optional;
@@ -692,6 +693,12 @@ public class DTOConverters {
}
Object value = ((Literal) expression).value();
+ // Binary literals use an explicit hexadecimal representation on the
string-valued wire API.
+ // byte[].toString() is an object identity and cannot preserve the
literal's bytes.
+ if (((Literal) expression).dataType() instanceof Types.BinaryType
+ && value instanceof byte[]) {
+ value = "0x" + BaseEncoding.base16().lowerCase().encode((byte[])
value);
+ }
return LiteralDTO.builder()
.withValue(value == null ? null : value.toString())
.withDataType(((Literal) expression).dataType())
diff --git
a/common/src/test/java/org/apache/gravitino/dto/util/TestDTOConverters.java
b/common/src/test/java/org/apache/gravitino/dto/util/TestDTOConverters.java
index 46a2ffa11b..f13ce8d7ad 100644
--- a/common/src/test/java/org/apache/gravitino/dto/util/TestDTOConverters.java
+++ b/common/src/test/java/org/apache/gravitino/dto/util/TestDTOConverters.java
@@ -19,6 +19,7 @@
package org.apache.gravitino.dto.util;
+import com.google.common.io.BaseEncoding;
import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
@@ -41,9 +42,27 @@ import org.apache.gravitino.rel.partitions.RangePartition;
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 TestDTOConverters {
+ @ParameterizedTest
+ @ValueSource(strings = {"4142", "0080ff", ""})
+ void testBinaryLiteralBytesUseHex(String hex) {
+ LiteralDTO dto =
+ (LiteralDTO)
+ DTOConverters.toFunctionArg(
+ Literals.of(BaseEncoding.base16().lowerCase().decode(hex),
Types.BinaryType.get()));
+ Assertions.assertEquals(Types.BinaryType.get(), dto.dataType());
+ Assertions.assertEquals("0x" + hex, dto.value());
+ Assertions.assertEquals("0x" + hex, ((Literal<?>)
DTOConverters.fromFunctionArg(dto)).value());
+ Assertions.assertEquals(
+ "legacy",
+ ((LiteralDTO) DTOConverters.toFunctionArg(Literals.of("legacy",
Types.BinaryType.get())))
+ .value());
+ }
+
@Test
void testIdentityPartitionDTOConvert() {
diff --git a/docs/expression.md b/docs/expression.md
index 8962497020..790a1a053e 100644
--- a/docs/expression.md
+++ b/docs/expression.md
@@ -46,6 +46,38 @@ NamedReference field = NamedReference.field("student");
## Literal
Literal is a constant value.
+
+Binary literals created from a Java `byte[]` are encoded in REST literal
values as `0x` followed by
+hexadecimal byte pairs, for example `"dataType": "binary", "value": "0x4142"`
for the bytes `AB`.
+`"0x"` represents an empty binary value. JDBC default converters recognize
this encoding and render
+the bytes using their dialect's binary syntax; provider restrictions on binary
defaults still apply.
+Unprefixed binary string values retain their previous provider-specific
interpretation. A binary
+string starting with `0x` must contain valid hexadecimal pairs; it is treated
as encoded bytes.
+Sending encoded binary defaults requires the server-side JDBC default
converters described here;
+older servers do not decode this representation.
+
+Spark timestamp defaults are converted to the target column type in the Spark
session time zone.
+An LTZ literal assigned to an NTZ column keeps the local fields produced by
Spark's timestamp cast;
+an NTZ literal assigned to an LTZ column is interpreted in the Spark session
zone before it is sent
+as an instant. DATE defaults assigned to timestamp columns become midnight in
the Spark session
+zone for LTZ targets, or local midnight for NTZ targets. String defaults for
timestamp columns are
+explicitly rejected in Spark 3.4 and 3.5; use a `TIMESTAMP_LTZ` or
`TIMESTAMP_NTZ` literal instead
+of relying on implicit database conversion. Date and string defaults for
non-timestamp columns
+retain their literal values and types.
+
+Timestamp, date, and string defaults load the current table schema once
through Gravitino and
+require table load permission, including date/string defaults for
non-timestamp columns. Other
+literal defaults do not add a schema read. Batches combining any of these
schema-dependent defaults
+with column additions, deletions, renames, or type changes are rejected before
writing because they
+require sequential schema evolution.
+
+JDBC timestamp defaults include seconds even for whole-minute and midnight
values.
+Zoned timestamp defaults use numeric UTC offsets in JDBC SQL. MySQL requires
8.0.19 or later
+for [datetime literals with
offsets](https://dev.mysql.com/doc/refman/8.0/en/datetime.html);
+MySQL 5.7 and older 8.0 releases cannot use these zoned defaults. Spark `DROP
DEFAULT` is not
+supported by the current Gravitino update request and is rejected explicitly;
`SET DEFAULT NULL`
+and `SET DEFAULT ''` remain distinct supported literal values.
+
The following is an example of creating a literal expression, demonstrating
how to create a `NULL` literal and three different data types of literal
expressions for the value `1024`.
<Tabs groupId='language' queryString>
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter.java
index 1db82c2d8c..06fef33be3 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter.java
@@ -19,6 +19,8 @@
package org.apache.gravitino.spark.connector;
import com.google.common.base.Preconditions;
+import java.util.function.Supplier;
+import org.apache.gravitino.rel.Column;
import org.apache.spark.sql.connector.catalog.TableChange;
public class SparkTableChangeConverter {
@@ -93,6 +95,21 @@ public class SparkTableChangeConverter {
}
}
+ /**
+ * Converts a table change with a lazy supplier for target column
information.
+ *
+ * <p>Version-specific converters may use the schema for defaults supported
by that Spark version.
+ * The Spark 3.3 converter does not load the schema.
+ *
+ * @param change the Spark table change
+ * @param columns supplies the current table columns only when required by
the change
+ * @return the converted Gravitino table change
+ */
+ public org.apache.gravitino.rel.TableChange toGravitinoTableChange(
+ TableChange change, Supplier<Column[]> columns) {
+ return toGravitinoTableChange(change);
+ }
+
private org.apache.gravitino.rel.TableChange.ColumnPosition
transformColumnPosition(
TableChange.ColumnPosition columnPosition) {
if (null == columnPosition) {
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
index 4492697486..90e2ff8ae1 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
@@ -20,11 +20,13 @@
package org.apache.gravitino.spark.connector.catalog;
import com.google.common.base.Preconditions;
+import com.google.common.base.Suppliers;
import com.google.common.collect.Sets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
+import java.util.function.Supplier;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
@@ -41,6 +43,7 @@ import org.apache.gravitino.function.Function;
import org.apache.gravitino.function.FunctionDefinition;
import org.apache.gravitino.function.FunctionImpl;
import org.apache.gravitino.function.JavaImpl;
+import org.apache.gravitino.rel.Column;
import org.apache.gravitino.rel.View;
import org.apache.gravitino.spark.connector.ConnectorConstants;
import org.apache.gravitino.spark.connector.PropertiesConverter;
@@ -274,11 +277,31 @@ public abstract class BaseCatalog implements
TableCatalog, SupportsNamespaces, F
@Override
public Table alterTable(Identifier ident, TableChange... changes) throws
NoSuchTableException {
- org.apache.gravitino.rel.TableChange[] gravitinoTableChanges =
- Arrays.stream(changes)
- .map(sparkTableChangeConverter::toGravitinoTableChange)
- .toArray(org.apache.gravitino.rel.TableChange[]::new);
try {
+ Supplier<Column[]> columns =
+ Suppliers.memoize(
+ () -> {
+ // A schema-changing batch needs sequential schema
evolution before coercion.
+ // Reject it before any write rather than resolve against
stale column types.
+ Preconditions.checkArgument(
+ Arrays.stream(changes)
+ .noneMatch(
+ change ->
+ change instanceof TableChange.AddColumn
+ || change instanceof
TableChange.DeleteColumn
+ || change instanceof
TableChange.RenameColumn
+ || change instanceof
TableChange.UpdateColumnType),
+ "Timestamp, date, and string defaults cannot be
combined with column schema changes");
+ return gravitinoCatalogClient
+ .asTableCatalog()
+ .loadTable(NameIdentifier.of(getDatabase(ident),
ident.name()))
+ .columns();
+ })
+ ::get;
+ org.apache.gravitino.rel.TableChange[] gravitinoTableChanges =
+ Arrays.stream(changes)
+ .map(change ->
sparkTableChangeConverter.toGravitinoTableChange(change, columns))
+ .toArray(org.apache.gravitino.rel.TableChange[]::new);
invalidateTable(ident);
org.apache.gravitino.rel.Table gravitinoTable =
gravitinoCatalogClient
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestTransformTableChange.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestTransformTableChange.java
index 523deb9021..1dc0e93970 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestTransformTableChange.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestTransformTableChange.java
@@ -32,6 +32,20 @@ public class TestTransformTableChange {
private SparkTableChangeConverter sparkTableChangeConverter =
new SparkTableChangeConverter(new SparkTypeConverter());
+ @Test
+ void testTransformWithoutLoadingTargetSchema() {
+ org.apache.gravitino.rel.TableChange change =
+ sparkTableChangeConverter.toGravitinoTableChange(
+ TableChange.setProperty("key", "value"),
+ () -> {
+ throw new AssertionError("Spark 3.3 table changes must not load
the schema");
+ });
+ org.apache.gravitino.rel.TableChange.SetProperty property =
+
Assertions.assertInstanceOf(org.apache.gravitino.rel.TableChange.SetProperty.class,
change);
+ Assertions.assertEquals("key", property.getProperty());
+ Assertions.assertEquals("value", property.getValue());
+ }
+
@Test
void testTransformSetProperty() {
TableChange sparkSetProperty = TableChange.setProperty("key", "value");
diff --git a/spark-connector/v3.4/spark/build.gradle.kts
b/spark-connector/v3.4/spark/build.gradle.kts
index d4da6062fc..bd3dcf840f 100644
--- a/spark-connector/v3.4/spark/build.gradle.kts
+++ b/spark-connector/v3.4/spark/build.gradle.kts
@@ -143,6 +143,7 @@ dependencies {
}
testImplementation(libs.junit.jupiter.api)
testImplementation(libs.junit.jupiter.params)
+ testImplementation(libs.mockito.core)
testImplementation(libs.mysql.driver)
testImplementation(libs.postgresql.driver)
testImplementation(libs.testcontainers)
diff --git
a/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter34.java
b/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter34.java
index 2aa8b1a807..fa95206864 100644
---
a/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter34.java
+++
b/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/SparkTableChangeConverter34.java
@@ -18,24 +18,157 @@
*/
package org.apache.gravitino.spark.connector;
+import com.google.common.base.Preconditions;
+import java.time.LocalDate;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.util.Arrays;
+import java.util.function.Supplier;
+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.Type;
+import org.apache.gravitino.rel.types.Types;
+import org.apache.gravitino.spark.connector.utils.SparkPartitionUtils;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.catalyst.expressions.Literal;
+import org.apache.spark.sql.catalyst.expressions.UnaryMinus;
+import org.apache.spark.sql.catalyst.expressions.UnaryPositive;
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser$;
+import org.apache.spark.sql.catalyst.util.DateTimeUtils;
import org.apache.spark.sql.connector.catalog.TableChange;
+import org.apache.spark.sql.internal.SQLConf;
+import org.apache.spark.sql.types.BinaryType;
+import org.apache.spark.sql.types.DateType;
+import org.apache.spark.sql.types.StringType;
+import org.apache.spark.sql.types.TimestampNTZType;
+import org.apache.spark.sql.types.TimestampType;
public class SparkTableChangeConverter34 extends SparkTableChangeConverter {
public SparkTableChangeConverter34(SparkTypeConverter sparkTypeConverter) {
super(sparkTypeConverter);
}
+ /** {@inheritDoc} */
@Override
public org.apache.gravitino.rel.TableChange
toGravitinoTableChange(TableChange change) {
+ return toGravitinoTableChange(
+ change,
+ () -> {
+ throw new IllegalArgumentException(
+ "Timestamp, date, and string defaults require the target column
schema");
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public org.apache.gravitino.rel.TableChange toGravitinoTableChange(
+ TableChange change, Supplier<Column[]> columns) {
if (change instanceof TableChange.UpdateColumnDefaultValue) {
- TableChange.UpdateColumnDefaultValue updateColumnDefaultValue =
+ TableChange.UpdateColumnDefaultValue defaultValue =
(TableChange.UpdateColumnDefaultValue) change;
return org.apache.gravitino.rel.TableChange.updateColumnDefaultValue(
- updateColumnDefaultValue.fieldNames(),
- Literals.stringLiteral(updateColumnDefaultValue.newDefaultValue()));
- } else {
- return super.toGravitinoTableChange(change);
+ defaultValue.fieldNames(),
+ convertDefaultValue(defaultValue.newDefaultValue(),
defaultValue.fieldNames(), columns));
+ }
+ return super.toGravitinoTableChange(change);
+ }
+
+ private Expression convertDefaultValue(
+ String sql, String[] fieldNames, Supplier<Column[]> columns) {
+ // Spark uses an empty SQL string for DROP DEFAULT; a string literal
containing no characters
+ // is represented by two quote characters instead.
+ if (sql.isEmpty()) {
+ throw new UnsupportedOperationException("DROP DEFAULT is not supported
by Gravitino");
+ }
+ org.apache.spark.sql.catalyst.expressions.Expression parsed;
+ Object value;
+ try {
+ parsed = CatalystSqlParser$.MODULE$.parseExpression(sql);
+ boolean literal =
+ parsed instanceof Literal
+ || (parsed instanceof UnaryMinus && ((UnaryMinus)
parsed).child() instanceof Literal)
+ || (parsed instanceof UnaryPositive
+ && ((UnaryPositive) parsed).child() instanceof Literal);
+ Preconditions.checkArgument(
+ literal && parsed.resolved() && parsed.foldable(),
+ "Only literal column defaults are supported: %s",
+ sql);
+ value = parsed.eval(null);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Unsupported column default SQL: " +
sql, e);
+ }
+ boolean timestampLiteral =
+ parsed.dataType() instanceof TimestampType || parsed.dataType()
instanceof TimestampNTZType;
+ if (timestampLiteral
+ || parsed.dataType() instanceof DateType
+ || parsed.dataType() instanceof StringType) {
+ Type target = columnType(columns.get(), fieldNames);
+ Preconditions.checkArgument(
+ !timestampLiteral || target instanceof Types.TimestampType,
+ "Timestamp defaults require a timestamp target column: %s",
+ String.join(".", fieldNames));
+ if (target instanceof Types.TimestampType) {
+ Preconditions.checkArgument(
+ !(parsed.dataType() instanceof StringType),
+ "String defaults for timestamp column %s are unsupported; use a
TIMESTAMP_LTZ or TIMESTAMP_NTZ literal",
+ String.join(".", fieldNames));
+ // Coerce in Spark's session zone before JDBC sees the value. A
database connection's
+ // zone must not decide how a date or local timestamp is assigned to
an instant.
+ ZoneId zone =
DateTimeUtils.getZoneId(SQLConf.get().sessionLocalTimeZone());
+ boolean targetZoned = ((Types.TimestampType) target).hasTimeZone();
+ if (parsed.dataType() instanceof DateType) {
+ return targetZoned
+ ? Literals.of(
+
DateTimeUtils.microsToInstant(DateTimeUtils.daysToMicros((int) value, zone))
+ .atOffset(ZoneOffset.UTC),
+ Types.TimestampType.withTimeZone())
+ : Literals.timestampLiteral(LocalDate.ofEpochDay((int)
value).atStartOfDay());
+ }
+ boolean sourceZoned = parsed.dataType() instanceof TimestampType;
+ if (targetZoned) {
+ return Literals.of(
+ (sourceZoned
+ ? DateTimeUtils.microsToInstant((long) value)
+ : DateTimeUtils.microsToLocalDateTime((long)
value).atZone(zone).toInstant())
+ .atOffset(ZoneOffset.UTC),
+ Types.TimestampType.withTimeZone());
+ }
+ return Literals.timestampLiteral(
+ sourceZoned
+ ? DateTimeUtils.microsToInstant((long)
value).atZone(zone).toLocalDateTime()
+ : DateTimeUtils.microsToLocalDateTime((long) value));
+ }
+ }
+ if (parsed.dataType() instanceof BinaryType) {
+ return Literals.of((byte[]) value, Types.BinaryType.get());
+ }
+ return SparkPartitionUtils.toGravitinoLiteral(
+ new GenericInternalRow(new Object[] {value}), 0, parsed.dataType());
+ }
+
+ private Type columnType(Column[] columns, String[] fieldNames) {
+ Type type =
+ Arrays.stream(columns)
+ .filter(column -> column.name().equals(fieldNames[0]))
+ .findFirst()
+ .orElseThrow(
+ () -> new IllegalArgumentException("Unknown default column: "
+ fieldNames[0]))
+ .dataType();
+ for (int i = 1; i < fieldNames.length; i++) {
+ Preconditions.checkArgument(
+ type instanceof Types.StructType,
+ "Unsupported nested default column: %s",
+ String.join(".", fieldNames));
+ String fieldName = fieldNames[i];
+ type =
+ Arrays.stream(((Types.StructType) type).fields())
+ .filter(field -> field.name().equals(fieldName))
+ .findFirst()
+ .orElseThrow(
+ () -> new IllegalArgumentException("Unknown default column:
" + fieldName))
+ .type();
}
+ return type;
}
}
diff --git
a/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/TestSparkTableChangeConverter34.java
b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/TestSparkTableChangeConverter34.java
index 67ea6c7bbd..e8b77f24b7 100644
---
a/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/TestSparkTableChangeConverter34.java
+++
b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/TestSparkTableChangeConverter34.java
@@ -16,39 +16,396 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.gravitino.spark.connector;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.util.HexFormat;
+import java.util.Locale;
+import
org.apache.gravitino.catalog.jdbc.converter.JdbcColumnDefaultValueConverter;
+import org.apache.gravitino.dto.rel.expressions.LiteralDTO;
+import
org.apache.gravitino.dto.requests.TableUpdateRequest.UpdateTableColumnDefaultValueRequest;
+import org.apache.gravitino.dto.util.DTOConverters;
+import org.apache.gravitino.json.JsonUtils;
+import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.expressions.Expression;
+import org.apache.gravitino.rel.expressions.literals.Literal;
import org.apache.gravitino.rel.expressions.literals.Literals;
+import org.apache.gravitino.rel.types.Decimal;
+import org.apache.gravitino.rel.types.Types;
+import org.apache.spark.sql.catalyst.expressions.Cast;
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser$;
+import org.apache.spark.sql.catalyst.util.DateTimeUtils;
import org.apache.spark.sql.connector.catalog.TableChange;
+import org.apache.spark.sql.internal.SQLConf;
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 TestSparkTableChangeConverter34 {
-
private SparkTableChangeConverter sparkTableChangeConverter =
new SparkTableChangeConverter34(new SparkTypeConverter34());
@Test
- void testUpdateColumnDefaultValue() {
+ void testTransformSetProperty() {
+ TableChange sparkSetProperty = TableChange.setProperty("key", "value");
+ org.apache.gravitino.rel.TableChange tableChange =
+ sparkTableChangeConverter.toGravitinoTableChange(sparkSetProperty);
+ Assertions.assertTrue(tableChange instanceof
org.apache.gravitino.rel.TableChange.SetProperty);
+ org.apache.gravitino.rel.TableChange.SetProperty gravitinoSetProperty =
+ (org.apache.gravitino.rel.TableChange.SetProperty) tableChange;
+ Assertions.assertEquals("key", gravitinoSetProperty.getProperty());
+ Assertions.assertEquals("value", gravitinoSetProperty.getValue());
+ }
+
+ @Test
+ void testTransformUpdateColumnDefaultValue() {
String[] fieldNames = new String[] {"col"};
- String defaultValue = "col_default_value";
- TableChange.UpdateColumnDefaultValue sparkUpdateColumnDefaultValue =
+ String defaultValue = "'col_default_value'";
+ TableChange.UpdateColumnDefaultValue sparkChange =
(TableChange.UpdateColumnDefaultValue)
TableChange.updateColumnDefaultValue(fieldNames, defaultValue);
org.apache.gravitino.rel.TableChange gravitinoChange =
-
sparkTableChangeConverter.toGravitinoTableChange(sparkUpdateColumnDefaultValue);
+ sparkTableChangeConverter.toGravitinoTableChange(
+ sparkChange, () -> new Column[] {Column.of("col",
Types.StringType.get(), null)});
+
+ Assertions.assertInstanceOf(
+ org.apache.gravitino.rel.TableChange.UpdateColumnDefaultValue.class,
gravitinoChange);
+ org.apache.gravitino.rel.TableChange.UpdateColumnDefaultValue converted =
+ (org.apache.gravitino.rel.TableChange.UpdateColumnDefaultValue)
gravitinoChange;
+ Assertions.assertArrayEquals(sparkChange.fieldNames(),
converted.fieldName());
+ Assertions.assertEquals(
+ Literals.stringLiteral("col_default_value"),
converted.getNewDefaultValue());
+ }
+
+ @Test
+ void testDefaultSqlIsConvertedToTypedLiterals() {
+ assertDefault("42", Literals.integerLiteral(42));
+ assertDefault("-42", Literals.integerLiteral(-42));
+ assertDefault("12.34", Literals.decimalLiteral(Decimal.of("12.34", 4, 2)));
+ assertDefault("true", Literals.booleanLiteral(true));
+ assertDefault("NULL", Literals.NULL);
+ assertDefault("''", Literals.stringLiteral(""));
+ assertDefault("DATE '2026-11-04'", Literals.dateLiteral("2026-11-04"));
+ UnsupportedOperationException exception =
+ Assertions.assertThrows(
+ UnsupportedOperationException.class,
+ () ->
+ sparkTableChangeConverter.toGravitinoTableChange(
+ TableChange.updateColumnDefaultValue(new String[] {"col"},
"")));
+ Assertions.assertEquals("DROP DEFAULT is not supported by Gravitino",
exception.getMessage());
+ for (String expression :
+ new String[] {
+ "1 + 1",
+ "current_timestamp()",
+ "upper('abc')",
+ "not valid SQL!",
+ "column_name",
+ "+(1 + 1)",
+ "+'abc'",
+ "+true"
+ }) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ sparkTableChangeConverter.toGravitinoTableChange(
+ TableChange.updateColumnDefaultValue(new String[] {"col"},
expression)));
+ }
+ }
+
+ @Test
+ void testExplicitPositiveLiteralDefaults() {
+ assertDefault("+42", Literals.integerLiteral(42));
+ assertDefault("+42L", Literals.longLiteral(42L));
+ assertDefault("+12.34", Literals.decimalLiteral(Decimal.of("12.34", 4,
2)));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"UTC", "Asia/Shanghai", "America/Los_Angeles"})
+ void testTimestampDefaultsPreserveInstant(String timeZone) {
+ withSessionConfig(
+ timeZone,
+ "TIMESTAMP_LTZ",
+ () -> {
+ for (String text :
+ new String[] {
+ "2026-11-04T12:34:56.123456",
+ "1969-12-31T23:59:59.999999",
+ "2026-11-04T12:34:00",
+ "2026-11-04T00:00:00"
+ }) {
+ Expression expected =
+ Literals.of(
+ LocalDateTime.parse(text)
+ .atZone(ZoneId.of(timeZone))
+ .toInstant()
+ .atOffset(ZoneOffset.UTC),
+ Types.TimestampType.withTimeZone());
+ Expression converted = assertDefault("TIMESTAMP '" + text + "'",
expected);
+ String expectedSql =
+ "'"
+ + ((Literal<?>) expected)
+ .value()
+ .toString()
+ .replace("T", " ")
+ .replace("Z", "+00:00")
+ .replaceFirst("^(\\d{4}-\\d{2}-\\d{2}
\\d{2}:\\d{2})(?=[+-])", "$1:00")
+ + "'";
+ Assertions.assertEquals(
+ expectedSql,
+ new JdbcColumnDefaultValueConverter()
+ .fromGravitino(DTOConverters.toFunctionArg(converted)));
+ assertDefault("TIMESTAMP_LTZ '" + text + "'", expected);
+ }
+ assertDefault(
+ "TIMESTAMP '2026-11-04 12:34:56.123456+02:00'",
+ Literals.of(
+
LocalDateTime.parse("2026-11-04T10:34:56.123456").atOffset(ZoneOffset.UTC),
+ Types.TimestampType.withTimeZone()));
+ });
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"UTC", "Asia/Shanghai", "America/Los_Angeles"})
+ void testTimestampNtzDefaultsPreserveLocalTime(String timeZone) {
+ withSessionConfig(
+ timeZone,
+ "TIMESTAMP_NTZ",
+ () -> {
+ for (String text :
+ new String[] {
+ "2026-11-04T12:34:56.123456",
+ "1969-12-31T23:59:59.999999",
+ "2026-11-04T12:34:00",
+ "2026-11-04T00:00:00"
+ }) {
+ Expression expected = Literals.timestampLiteral(text);
+ assertDefault("TIMESTAMP_NTZ '" + text + "'", expected);
+ assertDefault("TIMESTAMP '" + text + "'", expected);
+ }
+ });
+ }
- Assertions.assertTrue(
- gravitinoChange instanceof
org.apache.gravitino.rel.TableChange.UpdateColumnDefaultValue);
- org.apache.gravitino.rel.TableChange.UpdateColumnDefaultValue
- gravitinoUpdateColumnDefaultValue =
- (org.apache.gravitino.rel.TableChange.UpdateColumnDefaultValue)
gravitinoChange;
+ @ParameterizedTest
+ @ValueSource(strings = {"UTC", "Asia/Shanghai", "America/Los_Angeles"})
+ void testTimestampDefaultsAreCoercedToTargetColumn(String timeZone) {
+ for (String timestampType : new String[] {"TIMESTAMP_LTZ",
"TIMESTAMP_NTZ"}) {
+ withSessionConfig(
+ timeZone,
+ timestampType,
+ () -> {
+ String text = "2026-11-04T12:34:56.123456";
+ assertDefault("TIMESTAMP_LTZ '" + text + "'",
Literals.timestampLiteral(text));
+ try {
+ Cast cast =
+ (Cast)
+ CatalystSqlParser$.MODULE$.parseExpression(
+ "CAST(TIMESTAMP_LTZ '" + text + "' AS
TIMESTAMP_NTZ)");
+ Assertions.assertEquals(
+ LocalDateTime.parse(text),
+ DateTimeUtils.microsToLocalDateTime(
+ (long) ((Cast) cast.withTimeZone(timeZone)).eval(null)));
+ } catch (Exception e) {
+ throw new AssertionError(e);
+ }
+ assertDefault(
+ "TIMESTAMP_NTZ '" + text + "'",
+ Literals.of(
+ LocalDateTime.parse(text)
+ .atZone(ZoneId.of(timeZone))
+ .toInstant()
+ .atOffset(ZoneOffset.UTC),
+ Types.TimestampType.withTimeZone()));
+ assertDefault(
+ "TIMESTAMP_LTZ '2026-11-04 12:34:56.123456+02:00'",
+ Literals.timestampLiteral(
+ LocalDateTime.parse(text)
+ .atOffset(ZoneOffset.ofHours(2))
+ .toInstant()
+ .atZone(ZoneId.of(timeZone))
+ .toLocalDateTime()));
+ });
+ }
+ }
- Assertions.assertArrayEquals(
- sparkUpdateColumnDefaultValue.fieldNames(),
gravitinoUpdateColumnDefaultValue.fieldName());
+ @ParameterizedTest
+ @ValueSource(strings = {"UTC", "Asia/Shanghai", "America/Los_Angeles",
"Pacific/Apia"})
+ void testDateDefaultsAreCoercedToTimestampTarget(String timeZone) {
+ for (String timestampType : new String[] {"TIMESTAMP_LTZ",
"TIMESTAMP_NTZ"}) {
+ withSessionConfig(
+ timeZone,
+ timestampType,
+ () -> {
+ for (String date : new String[] {"2026-11-04", "1969-12-31",
"2011-12-30"}) {
+ String sql = "DATE '" + date + "'";
+ for (String target : new String[] {"TIMESTAMP_LTZ",
"TIMESTAMP_NTZ"}) {
+ try {
+ Cast cast =
+ (Cast)
+ CatalystSqlParser$.MODULE$.parseExpression(
+ "CAST(" + sql + " AS " + target + ")");
+ long micros = (long) ((Cast)
cast.withTimeZone(timeZone)).eval(null);
+ Expression expected =
+ target.equals("TIMESTAMP_LTZ")
+ ? Literals.of(
+
DateTimeUtils.microsToInstant(micros).atOffset(ZoneOffset.UTC),
+ Types.TimestampType.withTimeZone())
+ :
Literals.timestampLiteral(DateTimeUtils.microsToLocalDateTime(micros));
+ Expression converted = assertDefault(sql, expected);
+ UpdateTableColumnDefaultValueRequest request =
+ new UpdateTableColumnDefaultValueRequest(
+ new String[] {"col"},
DTOConverters.toFunctionArg(converted));
+ UpdateTableColumnDefaultValueRequest restored =
+ JsonUtils.objectMapper()
+ .readValue(
+
JsonUtils.objectMapper().writeValueAsString(request),
+ UpdateTableColumnDefaultValueRequest.class);
+ String expectedSql =
+ target.equals("TIMESTAMP_LTZ")
+ ? "'"
+ + LocalDate.parse(date)
+ .atStartOfDay(ZoneId.of(timeZone))
+ .toInstant()
+ .atOffset(ZoneOffset.UTC)
+ .toString()
+ .replace("T", " ")
+ .replace("Z", "+00:00")
+ .replaceFirst(
+ "^(\\d{4}-\\d{2}-\\d{2}
\\d{2}:\\d{2})(?=[+-])", "$1:00")
+ + "'"
+ : "'" + date + " 00:00:00'";
+ Assertions.assertEquals(
+ expectedSql,
+ new JdbcColumnDefaultValueConverter()
+ .fromGravitino(restored.getNewDefaultValue()));
+ } catch (Exception e) {
+ throw new AssertionError(e);
+ }
+ }
+ }
+ });
+ }
+ }
+
+ @Test
+ void testStringDefaultsRejectTimestampTargets() {
+ for (Types.TimestampType target :
+ new Types.TimestampType[] {
+ Types.TimestampType.withTimeZone(),
Types.TimestampType.withoutTimeZone()
+ }) {
+ for (String sql :
+ new String[] {"'2026-11-04'", "'2026-11-04 12:34:56.123456'",
"'now'", "''"}) {
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ sparkTableChangeConverter.toGravitinoTableChange(
+ TableChange.updateColumnDefaultValue(new String[]
{"col"}, sql),
+ () -> new Column[] {Column.of("col", target, null)}));
+ Assertions.assertTrue(error.getMessage().contains("col"));
+ Assertions.assertTrue(error.getMessage().contains("TIMESTAMP_LTZ or
TIMESTAMP_NTZ"));
+ }
+ }
+ assertDefault(
+ "'2026-11-04 12:34:56.123456'", Literals.stringLiteral("2026-11-04
12:34:56.123456"));
+ assertDefault("''", Literals.stringLiteral(""));
+ assertDefault("DATE '2026-11-04'", Literals.dateLiteral("2026-11-04"));
+ }
+
+ @Test
+ void testNestedTimestampTargetAndInvalidColumnPaths() {
+ Column[] columns =
+ new Column[] {
+ Column.of(
+ "parent",
+ Types.StructType.of(
+ Types.StructType.Field.nullableField(
+ "ts", Types.TimestampType.withoutTimeZone())),
+ null)
+ };
+ org.apache.gravitino.rel.TableChange.UpdateColumnDefaultValue converted =
+ (org.apache.gravitino.rel.TableChange.UpdateColumnDefaultValue)
+ sparkTableChangeConverter.toGravitinoTableChange(
+ TableChange.updateColumnDefaultValue(
+ new String[] {"parent", "ts"}, "TIMESTAMP_NTZ '2026-11-04
12:34:00'"),
+ () -> columns);
Assertions.assertEquals(
- Literals.stringLiteral(defaultValue),
- gravitinoUpdateColumnDefaultValue.getNewDefaultValue());
+ Literals.timestampLiteral("2026-11-04T12:34:00"),
converted.getNewDefaultValue());
+ for (String[] path :
+ new String[][] {{"missing"}, {"parent", "missing"}, {"parent", "ts",
"child"}}) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ sparkTableChangeConverter.toGravitinoTableChange(
+ TableChange.updateColumnDefaultValue(path, "TIMESTAMP_NTZ
'2026-11-04 12:34:00'"),
+ () -> columns));
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"TIMESTAMP '2026-11-04 12:34:56'", "DATE
'2026-11-04'", "'2026-11-04'"})
+ void testTemporalDefaultRequiresTargetSchema(String sql) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ sparkTableChangeConverter.toGravitinoTableChange(
+ TableChange.updateColumnDefaultValue(new String[] {"col"},
sql)));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> assertDefault("TIMESTAMP '2026-11-04 12:34:56'",
Literals.stringLiteral("unused")));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"4142", "0080FF", ""})
+ void testBinaryLiteralDefaults(String hex) throws Exception {
+ Expression converted =
+ assertDefault(
+ "X'" + hex + "'", Literals.of(HexFormat.of().parseHex(hex),
Types.BinaryType.get()));
+ LiteralDTO dto = (LiteralDTO) DTOConverters.toFunctionArg(converted);
+ Assertions.assertEquals("0x" + hex.toLowerCase(Locale.ROOT), dto.value());
+ UpdateTableColumnDefaultValueRequest request =
+ new UpdateTableColumnDefaultValueRequest(new String[] {"col"}, dto);
+ UpdateTableColumnDefaultValueRequest restored =
+ JsonUtils.objectMapper()
+ .readValue(
+ JsonUtils.objectMapper().writeValueAsString(request),
+ UpdateTableColumnDefaultValueRequest.class);
+ Assertions.assertEquals(
+ "X'" + hex.toLowerCase(Locale.ROOT) + "'",
+ new
JdbcColumnDefaultValueConverter().fromGravitino(restored.getNewDefaultValue()));
+ }
+
+ private Expression assertDefault(String sql, Expression expected) {
+ org.apache.gravitino.rel.TableChange.UpdateColumnDefaultValue converted =
+ (org.apache.gravitino.rel.TableChange.UpdateColumnDefaultValue)
+ sparkTableChangeConverter.toGravitinoTableChange(
+ TableChange.updateColumnDefaultValue(new String[] {"col"},
sql),
+ () -> new Column[] {Column.of("col", ((Literal<?>)
expected).dataType(), null)});
+ Literal<?> expectedLiteral = (Literal<?>) expected;
+ if (expectedLiteral.value() instanceof byte[]) {
+ Literal<?> actual = (Literal<?>) converted.getNewDefaultValue();
+ Assertions.assertEquals(expectedLiteral.dataType(), actual.dataType());
+ Assertions.assertArrayEquals((byte[]) expectedLiteral.value(), (byte[])
actual.value());
+ } else {
+ Assertions.assertEquals(expected, converted.getNewDefaultValue());
+ }
+ return converted.getNewDefaultValue();
+ }
+
+ private void withSessionConfig(String timeZone, String timestampType,
Runnable assertions) {
+ SQLConf conf = new SQLConf();
+ conf.setConfString("spark.sql.session.timeZone", timeZone);
+ conf.setConfString("spark.sql.timestampType", timestampType);
+ SQLConf.withExistingConf(
+ conf,
+ () -> {
+ assertions.run();
+ return null;
+ });
}
}
diff --git
a/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/catalog/TestBaseCatalogDefaults34.java
b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/catalog/TestBaseCatalogDefaults34.java
new file mode 100644
index 0000000000..a1c3b0d956
--- /dev/null
+++
b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/catalog/TestBaseCatalogDefaults34.java
@@ -0,0 +1,284 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.spark.connector.catalog;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.time.OffsetDateTime;
+import java.util.Map;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.client.GravitinoClient;
+import org.apache.gravitino.exceptions.ForbiddenException;
+import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.TableCatalog;
+import org.apache.gravitino.rel.expressions.literals.Literals;
+import org.apache.gravitino.rel.types.Types;
+import org.apache.gravitino.spark.connector.PropertiesConverter;
+import org.apache.gravitino.spark.connector.SparkTableChangeConverter34;
+import org.apache.gravitino.spark.connector.SparkTransformConverter;
+import org.apache.gravitino.spark.connector.SparkTypeConverter;
+import org.apache.gravitino.spark.connector.SparkTypeConverter34;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.catalog.TableChange;
+import org.apache.spark.sql.internal.SQLConf;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+/** Tests schema loading and typed defaults on Spark 3.4 and later. */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class TestBaseCatalogDefaults34 {
+
+ private Catalog gravitinoCatalog;
+ private TableCatalog gravitinoTableCatalog;
+ private org.apache.spark.sql.connector.catalog.TableCatalog sparkCatalog;
+ private TestCatalog catalog;
+
+ @BeforeAll
+ void initCatalogManager() {
+ GravitinoCatalogManager.create("user", () -> mock(GravitinoClient.class));
+ }
+
+ @AfterAll
+ void cleanupCatalogManager() {
+ GravitinoCatalogManager.get().close();
+ }
+
+ @BeforeEach
+ void setUp() throws Exception {
+ gravitinoCatalog = mock(Catalog.class);
+ gravitinoTableCatalog = mock(TableCatalog.class);
+ sparkCatalog =
mock(org.apache.spark.sql.connector.catalog.TableCatalog.class);
+ when(gravitinoCatalog.asTableCatalog()).thenReturn(gravitinoTableCatalog);
+ catalog = new TestCatalog(gravitinoCatalog, sparkCatalog);
+ Field converter =
BaseCatalog.class.getDeclaredField("sparkTableChangeConverter");
+ converter.setAccessible(true);
+ converter.set(catalog, new SparkTableChangeConverter34(new
SparkTypeConverter34()));
+ }
+
+ @Test
+ void testAlterTimestampDefaultLoadsTargetSchemaBeforeWriting() throws
Exception {
+ Identifier identifier = Identifier.of(new String[] {"schema"}, "table_a");
+ NameIdentifier gravitinoIdentifier = NameIdentifier.of("schema",
"table_a");
+ org.apache.gravitino.rel.Table table =
mock(org.apache.gravitino.rel.Table.class);
+
when(gravitinoTableCatalog.loadTable(gravitinoIdentifier)).thenReturn(table);
+ when(table.columns())
+ .thenReturn(new Column[] {Column.of("ts",
Types.TimestampType.withoutTimeZone(6), null)});
+ Table sparkTable = mock(Table.class);
+ when(sparkCatalog.loadTable(identifier)).thenReturn(sparkTable);
+ SQLConf conf = new SQLConf();
+ conf.setConfString("spark.sql.session.timeZone", "Asia/Shanghai");
+ SQLConf.withExistingConf(
+ conf,
+ () -> {
+ Assertions.assertDoesNotThrow(
+ () ->
+ catalog.alterTable(
+ identifier,
+ TableChange.updateColumnDefaultValue(
+ new String[] {"ts"}, "TIMESTAMP_LTZ '2026-11-04
12:34:56.123456'")));
+ return null;
+ });
+ ArgumentCaptor<org.apache.gravitino.rel.TableChange> change =
+ ArgumentCaptor.forClass(org.apache.gravitino.rel.TableChange.class);
+ verify(gravitinoTableCatalog).alterTable(eq(gravitinoIdentifier),
change.capture());
+ Assertions.assertEquals(
+ Literals.timestampLiteral("2026-11-04T12:34:56.123456"),
+ ((org.apache.gravitino.rel.TableChange.UpdateColumnDefaultValue)
change.getValue())
+ .getNewDefaultValue());
+ verify(gravitinoTableCatalog).loadTable(gravitinoIdentifier);
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {"TIMESTAMP_LTZ '2026-11-04 12:34:00'", "DATE '2026-11-04'",
"'2026-11-04'"})
+ void testTemporalDefaultRejectsSchemaChangingBatchBeforeWriting(String sql) {
+ Identifier identifier = Identifier.of(new String[] {"schema"}, "table_a");
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ catalog.alterTable(
+ identifier,
+ TableChange.updateColumnDefaultValue(new String[] {"ts"},
sql),
+ TableChange.updateColumnType(new String[] {"ts"},
DataTypes.TimestampNTZType)));
+ Assertions.assertTrue(error.getMessage().contains("cannot be combined"));
+ Mockito.verifyNoInteractions(gravitinoTableCatalog, sparkCatalog);
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {"TIMESTAMP_LTZ '2026-11-04 12:34:00'", "DATE '2026-11-04'",
"'2026-11-04'"})
+ void testTemporalSchemaReadFailurePreventsWritingAndPreservesCause(String
sql) {
+ Identifier identifier = Identifier.of(new String[] {"schema"}, "table_a");
+ NameIdentifier gravitinoIdentifier = NameIdentifier.of("schema",
"table_a");
+ ForbiddenException denied = new ForbiddenException("denied schema read");
+
when(gravitinoTableCatalog.loadTable(gravitinoIdentifier)).thenThrow(denied);
+ Assertions.assertSame(
+ denied,
+ Assertions.assertThrows(
+ ForbiddenException.class,
+ () ->
+ catalog.alterTable(
+ identifier, TableChange.updateColumnDefaultValue(new
String[] {"ts"}, sql))));
+ verify(gravitinoTableCatalog).loadTable(gravitinoIdentifier);
+ Mockito.verifyNoMoreInteractions(gravitinoTableCatalog);
+ Mockito.verifyNoInteractions(sparkCatalog);
+ }
+
+ @Test
+ void testDateDefaultUsesSessionZoneBeforeWriting() throws Exception {
+ Identifier identifier = Identifier.of(new String[] {"schema"}, "table_a");
+ NameIdentifier gravitinoIdentifier = NameIdentifier.of("schema",
"table_a");
+ org.apache.gravitino.rel.Table table =
mock(org.apache.gravitino.rel.Table.class);
+
when(gravitinoTableCatalog.loadTable(gravitinoIdentifier)).thenReturn(table);
+ when(table.columns())
+ .thenReturn(new Column[] {Column.of("ts",
Types.TimestampType.withTimeZone(), null)});
+ SQLConf conf = new SQLConf();
+ conf.setConfString("spark.sql.session.timeZone", "Asia/Shanghai");
+ SQLConf.withExistingConf(
+ conf,
+ () -> {
+ Assertions.assertDoesNotThrow(
+ () ->
+ catalog.alterTable(
+ identifier,
+ TableChange.updateColumnDefaultValue(
+ new String[] {"ts"}, "DATE '2026-11-04'")));
+ return null;
+ });
+ ArgumentCaptor<org.apache.gravitino.rel.TableChange> change =
+ ArgumentCaptor.forClass(org.apache.gravitino.rel.TableChange.class);
+ verify(gravitinoTableCatalog).alterTable(eq(gravitinoIdentifier),
change.capture());
+ Assertions.assertEquals(
+ Literals.of(
+ OffsetDateTime.parse("2026-11-03T16:00:00Z"),
Types.TimestampType.withTimeZone()),
+ ((org.apache.gravitino.rel.TableChange.UpdateColumnDefaultValue)
change.getValue())
+ .getNewDefaultValue());
+ verify(gravitinoTableCatalog).loadTable(gravitinoIdentifier);
+ }
+
+ @Test
+ void testStringTimestampDefaultRejectedBeforeWriting() {
+ Identifier identifier = Identifier.of(new String[] {"schema"}, "table_a");
+ NameIdentifier gravitinoIdentifier = NameIdentifier.of("schema",
"table_a");
+ org.apache.gravitino.rel.Table table =
mock(org.apache.gravitino.rel.Table.class);
+
when(gravitinoTableCatalog.loadTable(gravitinoIdentifier)).thenReturn(table);
+ when(table.columns())
+ .thenReturn(new Column[] {Column.of("ts",
Types.TimestampType.withTimeZone(), null)});
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ catalog.alterTable(
+ identifier,
+ TableChange.updateColumnDefaultValue(
+ new String[] {"ts"}, "'2026-11-04 12:34:56.123456'")));
+ Assertions.assertTrue(error.getMessage().contains("TIMESTAMP_LTZ or
TIMESTAMP_NTZ"));
+ verify(gravitinoTableCatalog).loadTable(gravitinoIdentifier);
+ Mockito.verifyNoMoreInteractions(gravitinoTableCatalog);
+ Mockito.verifyNoInteractions(sparkCatalog);
+ }
+
+ @Test
+ void testNumericDefaultDoesNotReadSchema() throws Exception {
+ Identifier identifier = Identifier.of(new String[] {"schema"}, "table_a");
+ NameIdentifier gravitinoIdentifier = NameIdentifier.of("schema",
"table_a");
+ catalog.alterTable(identifier, TableChange.updateColumnDefaultValue(new
String[] {"ts"}, "42"));
+ verify(gravitinoTableCatalog,
Mockito.never()).loadTable(gravitinoIdentifier);
+ }
+
+ @Test
+ void testTemporalDefaultsShareOneSchemaRead() throws Exception {
+ Identifier identifier = Identifier.of(new String[] {"schema"}, "table_a");
+ NameIdentifier gravitinoIdentifier = NameIdentifier.of("schema",
"table_a");
+ org.apache.gravitino.rel.Table table =
mock(org.apache.gravitino.rel.Table.class);
+
when(gravitinoTableCatalog.loadTable(gravitinoIdentifier)).thenReturn(table);
+ when(table.columns())
+ .thenReturn(
+ new Column[] {
+ Column.of("ts", Types.TimestampType.withoutTimeZone(), null),
+ Column.of("label", Types.StringType.get(), null)
+ });
+ catalog.alterTable(
+ identifier,
+ TableChange.updateColumnDefaultValue(
+ new String[] {"ts"}, "TIMESTAMP_NTZ '2026-11-04 12:34:00'"),
+ TableChange.updateColumnDefaultValue(
+ new String[] {"ts"}, "TIMESTAMP_NTZ '2026-11-04 00:00:00'"),
+ TableChange.updateColumnDefaultValue(new String[] {"ts"}, "DATE
'2026-11-04'"),
+ TableChange.updateColumnDefaultValue(new String[] {"label"},
"'label'"));
+ verify(gravitinoTableCatalog).loadTable(gravitinoIdentifier);
+ verify(table).columns();
+ }
+
+ private static class TestCatalog extends BaseCatalog {
+
+ private TestCatalog(
+ Catalog gravitinoCatalog,
+ org.apache.spark.sql.connector.catalog.TableCatalog sparkCatalog) {
+ this.gravitinoCatalogClient = gravitinoCatalog;
+ this.sparkCatalog = sparkCatalog;
+ }
+
+ @Override
+ protected org.apache.spark.sql.connector.catalog.TableCatalog
createAndInitSparkCatalog(
+ String name, CaseInsensitiveStringMap options, Map<String, String>
properties) {
+ return sparkCatalog;
+ }
+
+ @Override
+ protected Table createSparkTable(
+ Identifier identifier,
+ org.apache.gravitino.rel.Table gravitinoTable,
+ Table sparkTable,
+ org.apache.spark.sql.connector.catalog.TableCatalog sparkCatalog,
+ PropertiesConverter propertiesConverter,
+ SparkTransformConverter sparkTransformConverter,
+ SparkTypeConverter sparkTypeConverter) {
+ return sparkTable;
+ }
+
+ @Override
+ protected PropertiesConverter getPropertiesConverter() {
+ return mock(PropertiesConverter.class);
+ }
+
+ @Override
+ protected SparkTransformConverter getSparkTransformConverter() {
+ return mock(SparkTransformConverter.class);
+ }
+ }
+}
