This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 1c1f3e51bb9 [test](external) Cover the external binary and timestamp
mapping cases from branch-4.2 (#68918)
1c1f3e51bb9 is described below
commit 1c1f3e51bb91b042dac53492a7bf501f4eadecf3
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Mon Oct 12 14:04:05 2026 +0800
[test](external) Cover the external binary and timestamp mapping cases from
branch-4.2 (#68918)
### What problem does this PR solve?
Issue Number: None
Related PR: #68532, #68786
Problem Summary:
**In short.** This ports the tests of branch-4.2 #68532 that master does
not have yet. Master re-implemented #68532 in #68786; a hunk-by-hunk
comparison found all of its product behavior on master, but about twenty
of its assertions had no master equivalent. This PR adds them, fixes
three Paimon tests that set a misspelled option, and changes no product
code.
**Background.** #68532 (merged into branch-4.1 and synced to branch-4.2)
maps external binary columns to VARBINARY and instant timestamps to
TIMESTAMPTZ across Hive, Iceberg, Paimon, JDBC, Hudi, MaxCompute and the
Trino connector. On master the external catalogs live in
`fe/fe-connector/*`, so #68786 re-implemented the change on the
connector plugins rather than cherry-picking it, and moved most of the
#68532 tests along. The ones listed below were not moved.
**The problem, and what it cost.** Without these tests nothing on master
fails if one of these behaviors regresses:
| Area | Behavior with no test on master |
| --- | --- |
| Iceberg partitions | 0x-hex BINARY/FIXED/UUID commit values are
decoded, and wrong lengths or malformed hex are rejected; transform
ranges carry `+00:00` bounds and the proleptic year zero; zoned and
unzoned partition timestamps round-trip in every session zone, including
both instants of a DST fold; commit values honor their explicit offsets
|
| Iceberg writes | zoned write defaults keep the instant and nested
binary defaults stay bytes with the old flags off; binary/UUID commit
values become typed partitions; a NULL static overwrite on a
binary/fixed/uuid key replaces only the NULL partition (full and hybrid
specs) and reaches the sink as a NULL marker distinct from bytes; MODIFY
to TIMESTAMPTZ on a zoned field is a no-op while DATETIMEV2 is rejected;
Iceberg does not use Hive's INT96 time zone |
| fe-core | TIMESTAMPTZ range partitions keep UTC bounds and a non-empty
NULL-min range in UTC, Asia/Shanghai and America/New_York sessions; the
legacy JDBC clients that CDC still uses map binary and instant types
regardless of the old flags, and PostgreSQL `timestamptz` becomes
nullable |
| Paimon | nested LTZ write columns stay TIMESTAMPTZ; LTZ partition
values keep pre-epoch and DST-fold instants in every zone; legacy ORC
LTZ files route to the SDK for MAP/MULTISET/ROW leaves and mixed splits
|
| JDBC / HMS / Trino | per-dialect binary and timestamp mappings;
calendar functions on instants, TIMESTAMPTZ literals, PostgreSQL decoded
NULLs and dialect-specific binary literals stay local; Hive binary
partition keys keep their text type; Trino domains encode UTC micros and
millisecond-aligned zoned bounds; the JDBC writer sets the MySQL session
to UTC before it prepares the INSERT |
| Regression | a Paimon LTZ INSERT SELECT of two instants that fold to
the same Los Angeles time; a Spark cross-check of LTZ rows written in a
non-UTC session, including a pre-epoch instant; a native V1 read of the
Paimon timestamp-scale tables |
Separately, three `PaimonScanPlanProviderTest` cases set
`orc.timestamp-ltz.legacy-type`; the real key is
`orc.timestamp-ltz.legacy.type`, so the setting was ignored and the
tests passed only because the default is `true`.
**How this PR fixes it.** It ports each missing #68532 assertion to the
master class that now owns the behavior (mostly connector tests),
adapted to the connector APIs; follows master's regression style
(`order_qt_` blocks plus Spark/Doris cross-checks instead of literal
asserts); and corrects the misspelled option key.
**Results.** The 19 changed test classes pass locally. To check that the
new tests are not vacuous, six product-code mutations were applied
across five modules: dropping the binary-literal pushdown guard, an
always-true NULL static overwrite filter, no FIXED length check, parsing
the NULL-min sentinel as a date, ignoring MULTISET when routing legacy
ORC LTZ, and initializing the MySQL UTC session after the prepare. Each
mutation failed exactly the corresponding new test and nothing else.
Both Paimon regression suites pass against the iceberg docker
environment, with the new `.out` blocks generated by the framework and
matching #68532's expected values (for example epoch micros
`1699173000123456` and `1699176600123456` for the fold).
Co-authored-by: Gabriel <[email protected]>
---
.../doris/jdbc/JdbcTimestampSemanticsTest.java | 32 ++++++
.../hms/ThriftHmsClientBinaryPartitionTest.java | 56 +++++++++
.../iceberg/IcebergConnectorTransactionTest.java | 60 ++++++++++
.../iceberg/IcebergNestedColumnEvolutionTest.java | 21 ++++
.../iceberg/IcebergPartitionUtilsTest.java | 93 +++++++++++++++
.../iceberg/IcebergScanPlanProviderTest.java | 7 ++
.../connector/iceberg/IcebergSchemaUtilsTest.java | 9 ++
.../iceberg/IcebergTimestampMappingTest.java | 31 +++--
.../iceberg/IcebergWritePlanProviderTest.java | 48 ++++++++
.../connector/iceberg/IcebergWriterHelperTest.java | 32 ++++++
.../jdbc/JdbcTimestampProjectionTest.java | 111 ++++++++++++++++++
.../jdbc/client/JdbcLogicalTypeMappingTest.java | 120 +++++++++++++++++++
.../paimon/PaimonLegacyOrcTimestampTest.java | 45 ++++++++
.../paimon/PaimonPartitionValueRenderTest.java | 19 +++
.../paimon/PaimonScanPlanProviderTest.java | 6 +-
.../paimon/PaimonWritePlanProviderTest.java | 21 ++++
.../connector/trino/TrinoTimestampMappingTest.java | 40 +++++++
.../datasource/jdbc/client/JdbcClientTest.java | 127 +++++++++++++++++++++
.../mvcc/PluginDrivenMvccExternalTableTest.java | 37 +++++-
.../paimon/paimon_timestamp_types.out | 6 +
.../paimon/write/test_paimon_write_types.out | 14 +++
.../paimon/paimon_timestamp_types.groovy | 5 +
.../paimon/write/test_paimon_write_types.groovy | 34 +++++-
23 files changed, 958 insertions(+), 16 deletions(-)
diff --git
a/fe/be-java-extensions/jdbc-scanner/src/test/java/org/apache/doris/jdbc/JdbcTimestampSemanticsTest.java
b/fe/be-java-extensions/jdbc-scanner/src/test/java/org/apache/doris/jdbc/JdbcTimestampSemanticsTest.java
index d7c97c0cbc3..6279a667008 100644
---
a/fe/be-java-extensions/jdbc-scanner/src/test/java/org/apache/doris/jdbc/JdbcTimestampSemanticsTest.java
+++
b/fe/be-java-extensions/jdbc-scanner/src/test/java/org/apache/doris/jdbc/JdbcTimestampSemanticsTest.java
@@ -106,6 +106,38 @@ class JdbcTimestampSemanticsTest {
order.verify(connection).prepareStatement("INSERT");
}
+ @Test
+ void
testMySqlWriterInitializesUtcBeforePreparingInsert(@org.junit.jupiter.api.io.TempDir
java.nio.file.Path dir)
+ throws Exception {
+ // Drive the production open path through the shared data-source cache
instead of a test-only seam.
+ String driverUrl =
java.nio.file.Files.createFile(dir.resolve("driver.jar")).toUri().toString();
+ java.util.Map<String, String> params = new java.util.HashMap<>();
+ params.put("table_type", "MYSQL");
+ params.put("jdbc_url", "jdbc:mysql://localhost/test");
+ params.put("jdbc_driver_url", driverUrl);
+ params.put("insert_sql", "INSERT INTO t VALUES (?)");
+ JdbcJniWriter writer = new JdbcJniWriter(1, params);
+ com.zaxxer.hikari.HikariDataSource dataSource =
Mockito.mock(com.zaxxer.hikari.HikariDataSource.class);
+ java.sql.Connection connection =
Mockito.mock(java.sql.Connection.class);
+ java.sql.Statement timezoneStatement =
Mockito.mock(java.sql.Statement.class);
+ Mockito.when(dataSource.getConnection()).thenReturn(connection);
+
Mockito.when(connection.createStatement()).thenReturn(timezoneStatement);
+ Mockito.when(connection.prepareStatement(Mockito.anyString()))
+ .thenReturn(Mockito.mock(java.sql.PreparedStatement.class));
+ String cacheKey = JdbcDataSource.createCacheKey(0L,
"jdbc:mysql://localhost/test", "", "", driverUrl,
+ "", "", 1, 10, 1800000, 5000, false);
+ JdbcDataSource.getDataSource().putSource(cacheKey, dataSource);
+ try {
+ writer.openInternal();
+ org.mockito.InOrder order = Mockito.inOrder(connection,
timezoneStatement);
+ order.verify(connection).createStatement();
+ order.verify(timezoneStatement).execute("SET SESSION time_zone =
'+00:00'");
+ order.verify(connection).prepareStatement("INSERT INTO t VALUES
(?)");
+ } finally {
+ JdbcDataSource.getDataSource().getSourcesMap().remove(cacheKey);
+ }
+ }
+
@Test
void testMySqlTimestampWriteUsesUtcSessionFields() throws Exception {
MySQLTypeHandler executor = new MySQLTypeHandler("MYSQL");
diff --git
a/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/ThriftHmsClientBinaryPartitionTest.java
b/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/ThriftHmsClientBinaryPartitionTest.java
new file mode 100644
index 00000000000..38f83fdfe03
--- /dev/null
+++
b/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/ThriftHmsClientBinaryPartitionTest.java
@@ -0,0 +1,56 @@
+// 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.doris.connector.hms;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Proxy;
+import java.util.Collections;
+
+class ThriftHmsClientBinaryPartitionTest {
+ @Test
+ void binaryPartitionKeyKeepsTextWhileDataColumnKeepsBytes() {
+ Table remote = new Table();
+ remote.setDbName("db");
+ remote.setTableName("tbl");
+ StorageDescriptor sd = new StorageDescriptor();
+ sd.setCols(Collections.singletonList(new FieldSchema("payload",
"binary", "")));
+ remote.setSd(sd);
+ remote.setPartitionKeys(Collections.singletonList(new
FieldSchema("part_key", "binary", "")));
+ IMetaStoreClient fake = (IMetaStoreClient) Proxy.newProxyInstance(
+ IMetaStoreClient.class.getClassLoader(), new Class<?>[]
{IMetaStoreClient.class},
+ (proxy, method, args) -> {
+ if ("getTable".equals(method.getName())) {
+ return remote;
+ }
+ return method.getReturnType() == boolean.class ? false :
null;
+ });
+ ThriftHmsClient client = new ThriftHmsClient(new
HmsClientConfig(Collections.emptyMap(), 0), null,
+ hiveConf -> fake, HmsTypeMapping.Options.DEFAULT);
+
+ HmsTableInfo table = client.getTable("db", "tbl");
+ // Partition values come from directory names, so existing partitions
must keep their text contract.
+ Assertions.assertEquals("STRING",
table.getPartitionKeys().get(0).getType().getTypeName());
+ Assertions.assertEquals("VARBINARY",
table.getColumns().get(0).getType().getTypeName());
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
index eb30afcddaf..35a4ae5ad29 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java
@@ -73,6 +73,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -907,6 +908,65 @@ public class IcebergConnectorTransactionTest {
"a stale static overwrite must fail at begin before it can
degrade to an always-true filter");
}
+ @Test
+ public void overwriteStaticNullBinaryKeyReplacesOnlyTheNullPartition() {
+ // A NULL static key must become key IS NULL for full and hybrid
specs; matching bytes or text
+ // instead would replace the wrong partition.
+ for (org.apache.iceberg.types.Type keyType : Arrays.asList(
+ Types.BinaryType.get(), Types.FixedType.ofLength(16),
Types.UUIDType.get())) {
+ for (boolean hybrid : new boolean[] {false, true}) {
+ InMemoryCatalog catalog = freshCatalog();
+ TableIdentifier id = TableIdentifier.of("db1", "t1");
+ Schema schema = new Schema(Types.NestedField.optional(1,
"key", keyType),
+ Types.NestedField.optional(2, "region",
Types.StringType.get()));
+ PartitionSpec.Builder builder =
PartitionSpec.builderFor(schema).identity("key");
+ if (hybrid) {
+ builder.identity("region");
+ }
+ PartitionSpec spec = builder.build();
+ Table table = catalog.createTable(id, schema, spec,
props("write.format.default", "parquet"));
+ Object nonNullKey = keyType.typeId() ==
org.apache.iceberg.types.Type.TypeID.UUID
+ ?
java.util.UUID.fromString("00112233-4455-6677-8899-aabbccddeeff")
+ : ByteBuffer.wrap(new byte[16]);
+ table.newAppend()
+ .appendFile(binaryKeyFile(spec,
"s3://b/db1/t1/null-key.parquet", null))
+ .appendFile(binaryKeyFile(spec,
"s3://b/db1/t1/non-null-key.parquet", nonNullKey))
+ .commit();
+
+ IcebergConnectorTransaction txn = txnFor(
+ opsReturning(catalog.loadTable(id)), new
RecordingConnectorContext());
+ txn.beginWrite(SESSION, "db1", "t1",
+ overwriteStaticCtx(table,
Collections.singletonMap("key", null)));
+
txn.addCommitData(commitBytes(dataFileItem("s3://b/db1/t1/replacement.parquet",
1L, 1024L,
+ hybrid ? Arrays.asList("null", "us") :
Collections.singletonList("null"))));
+ txn.commit();
+
+ Set<String> live = new HashSet<>();
+ for (DataFile file : currentDataFiles(catalog.loadTable(id))) {
+ live.add(file.path().toString());
+ }
+ Assertions.assertEquals(new HashSet<>(Arrays.asList(
+ "s3://b/db1/t1/non-null-key.parquet",
"s3://b/db1/t1/replacement.parquet")),
+ live, keyType + " hybrid=" + hybrid);
+ }
+ }
+ }
+
+ private static DataFile binaryKeyFile(PartitionSpec spec, String path,
Object key) {
+ org.apache.iceberg.PartitionData partition = new
org.apache.iceberg.PartitionData(spec.partitionType());
+ partition.set(0, key);
+ if (spec.fields().size() > 1) {
+ partition.set(1, "us");
+ }
+ return DataFiles.builder(spec)
+ .withPath(path)
+ .withFileSizeInBytes(1024)
+ .withRecordCount(1)
+ .withFormat(FileFormat.PARQUET)
+ .withPartition(partition)
+ .build();
+ }
+
@Test
public void overwriteStaticRejectsConcurrentDataInTargetPartition() {
InMemoryCatalog catalog = freshCatalog();
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolutionTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolutionTest.java
index c2029243c3a..6a069de0992 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolutionTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolutionTest.java
@@ -733,6 +733,27 @@ public class IcebergNestedColumnEvolutionTest {
Assertions.assertEquals(3, amount.scale());
}
+ @Test
+ public void
testModifyNestedTimestamptzKeepsInstantTypeAndRejectsWallClock() {
+ // Restating the mapped TIMESTAMPTZ type is a no-op; DATETIMEV2 would
drop the instant semantics.
+ Schema schema = new Schema(Types.NestedField.optional(1, "info",
Types.StructType.of(
+ Types.NestedField.optional(2, "tz_value",
Types.TimestampType.withZone(), "tz doc"))));
+ createTable("m_tz", schema);
+ ops.modifyNestedColumn("db1", "m_tz", path("info", "tz_value"),
+ change("tz_value",
IcebergSchemaBuilder.buildColumnType(ConnectorType.of("TIMESTAMPTZ", 6, 0)),
+ null, true), false, false, null);
+ Types.NestedField tzValue =
reload("m_tz").findField("info").type().asStructType().field("tz_value");
+ Assertions.assertEquals(Types.TimestampType.withZone(),
tzValue.type());
+ Assertions.assertEquals("tz doc", tzValue.doc());
+
+ assertFailsLoud(() -> ops.modifyNestedColumn("db1", "m_tz",
path("info", "tz_value"),
+ change("tz_value",
IcebergSchemaBuilder.buildColumnType(
+ ConnectorType.of("DATETIMEV2", 6, 0)), null,
true), false, false, null),
+ "Cannot change column type: info.tz_value");
+ Assertions.assertEquals(Types.TimestampType.withZone(),
+
reload("m_tz").findField("info").type().asStructType().field("tz_value").type());
+ }
+
@Test
public void testModifyNestedPrimitivePromotionDisallowedFailsLoud() {
// BIGINT -> INT is not an iceberg-representable promotion.
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsTest.java
index 28b95d18e56..323d28a73d8 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsTest.java
@@ -1353,4 +1353,97 @@ public class IcebergPartitionUtilsTest {
}
}
+ @Test
+ public void timestampPartitionsRoundTripInEverySessionZone() {
+ for (ZoneId zone : Arrays.asList(ZoneOffset.UTC, SHANGHAI,
ZoneId.of("America/New_York"))) {
+ for (Types.TimestampType type : Arrays.asList(
+ Types.TimestampType.withZone(),
Types.TimestampType.withoutZone())) {
+ // 1636263000123456 and 1636266600123456 are both
01:30:00.123456 in New York.
+ for (long micros : new long[] {-1L, -1_000_001L,
1636263000123456L, 1636266600123456L}) {
+ String encoded =
IcebergPartitionUtils.serializePartitionValue(type, micros, zone);
+ Assertions.assertEquals(micros,
IcebergPartitionUtils.parsePartitionValueFromString(
+ encoded, type, zone), type + " " + zone + " " +
micros);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void parseCommittedTimestampsHonorTheirExplicitOffsets() {
+ for (ZoneId zone : Arrays.asList(ZoneOffset.UTC, SHANGHAI,
ZoneId.of("America/New_York"))) {
+ // BE commits carry an offset, so the session zone must not
reinterpret either fold instant.
+ for (String value : Arrays.asList("2021-11-07
01:30:00.123456-04:00",
+ "2021-11-07 01:30:00.123456-05:00", "1969-12-31
23:59:59.999999+00:00")) {
+ java.time.Instant instant =
java.time.OffsetDateTime.parse(value.replace(' ', 'T')).toInstant();
+ long expected = instant.getEpochSecond() * 1_000_000L +
instant.getNano() / 1000;
+ Assertions.assertEquals(expected,
IcebergPartitionUtils.parsePartitionValueFromString(
+ value, Types.TimestampType.withZone(), zone), value +
" " + zone);
+ }
+ Assertions.assertEquals(-1L,
IcebergPartitionUtils.parsePartitionValueFromString(
+ "1969-12-31 23:59:59.999999",
Types.TimestampType.withoutZone(), zone));
+ }
+ }
+
+ @Test
+ public void parseBinaryAndUuidPartitionValuesFromCommits() {
+ byte[] bytes = new byte[] {0, (byte) 0xde, (byte) 0xad, (byte) 0xff};
+ for (org.apache.iceberg.types.Type type : Arrays.asList(
+ Types.BinaryType.get(),
Types.FixedType.ofLength(bytes.length))) {
+ Assertions.assertEquals(ByteBuffer.wrap(bytes),
IcebergPartitionUtils.parsePartitionValueFromString(
+ "0x00deadff", type, ZoneOffset.UTC));
+ Assertions.assertEquals("0x00deadff",
IcebergPartitionUtils.serializePartitionValue(
+ type, ByteBuffer.wrap(bytes), ZoneOffset.UTC));
+ }
+ Assertions.assertEquals(ByteBuffer.allocate(0),
IcebergPartitionUtils.parsePartitionValueFromString(
+ "0x", Types.BinaryType.get(), ZoneOffset.UTC));
+ UUID uuid = UUID.fromString("00112233-4455-6677-8899-aabbccddeeff");
+ Assertions.assertEquals(uuid,
IcebergPartitionUtils.parsePartitionValueFromString(
+ "0x00112233445566778899aabbccddeeff", Types.UUIDType.get(),
ZoneOffset.UTC));
+ Assertions.assertEquals(uuid,
IcebergPartitionUtils.parsePartitionValueFromString(
+ uuid.toString(), Types.UUIDType.get(), ZoneOffset.UTC));
+ // A wrong byte count or a value without the 0x prefix must not become
a different partition.
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
+ IcebergPartitionUtils.parsePartitionValueFromString(
+ "0xdead", Types.FixedType.ofLength(4),
ZoneOffset.UTC));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
+ IcebergPartitionUtils.parsePartitionValueFromString(
+ "0xdead", Types.UUIDType.get(), ZoneOffset.UTC));
+ for (String malformed : Arrays.asList("DEAD", "0xD", "0xGG")) {
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
+ IcebergPartitionUtils.parsePartitionValueFromString(
+ malformed, Types.BinaryType.get(),
ZoneOffset.UTC), malformed);
+ }
+ }
+
+ @Test
+ public void buildRangeTimestamptzBoundsCarryUtcOffset() {
+ // Transform ordinals describe UTC boundaries; the offset keeps
fe-core from reading them in the session zone.
+ Map<String, String> uppers = new HashMap<>();
+ uppers.put("hour", "1970-01-01 01:00:00+00:00");
+ uppers.put("day", "1970-01-02 00:00:00+00:00");
+ uppers.put("month", "1970-02-01 00:00:00+00:00");
+ uppers.put("year", "1971-01-01 00:00:00+00:00");
+ for (Map.Entry<String, String> entry : uppers.entrySet()) {
+ IcebergPartitionUtils.RangeBuild rb =
IcebergPartitionUtils.buildRange(
+ "ts_" + entry.getKey() + "=0", "0", entry.getKey(),
Types.TimestampType.withZone(), 1L, 1L);
+ Assertions.assertEquals(Collections.singletonList("1970-01-01
00:00:00+00:00"), rb.getLowerBound());
+
Assertions.assertEquals(Collections.singletonList(entry.getValue()),
rb.getUpperBound());
+ }
+ }
+
+ @Test
+ public void buildRangeYearZeroUsesProlepticYear() {
+ // Year-of-era formatting would render both year zero and year one as
0001.
+ Map<org.apache.iceberg.types.Type, String> suffixes = new HashMap<>();
+ suffixes.put(Types.DateType.get(), "");
+ suffixes.put(Types.TimestampType.withoutZone(), " 00:00:00");
+ suffixes.put(Types.TimestampType.withZone(), " 00:00:00+00:00");
+ for (Map.Entry<org.apache.iceberg.types.Type, String> entry :
suffixes.entrySet()) {
+ IcebergPartitionUtils.RangeBuild rb =
IcebergPartitionUtils.buildRange(
+ "y=-1970", "-1970", "year", entry.getKey(), 1L, 1L);
+ Assertions.assertEquals(Collections.singletonList("0000-01-01" +
entry.getValue()), rb.getLowerBound());
+ Assertions.assertEquals(Collections.singletonList("0001-01-01" +
entry.getValue()), rb.getUpperBound());
+ }
+ }
+
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
index 3e39b6aa47a..f0d616d7431 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
@@ -310,6 +310,13 @@ public class IcebergScanPlanProviderTest {
Assertions.assertTrue(provider.ignorePartitionPruneShortCircuit());
}
+ @Test
+ public void doesNotUseHiveParquetInt96TimeZone() {
+ // Iceberg INT96 values are UTC instants; Hive's session-zone rule
would shift them.
+ Assertions.assertFalse(new
IcebergScanPlanProvider(IcebergCatalogProperties.of(Collections.emptyMap()),
+ new
RecordingIcebergCatalogOps()).usesHiveParquetInt96TimeZone());
+ }
+
@Test
public void supportsSystemTableTimeTravelIsTrue() {
IcebergScanPlanProvider provider =
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
index 7d6c790b927..0e7e3173678 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
@@ -398,6 +398,15 @@ public class IcebergSchemaUtilsTest {
Assertions.assertEquals("key", nested.get(1).getFieldPtr().getName());
}
+ @Test
+ public void timestamptzInitialDefaultKeepsUtcOffsetWithoutLegacyFlag() {
+ // The offset lets BE read a missing column's default as an instant in
every session zone.
+ for (boolean flag : new boolean[] {false, true}) {
+ Assertions.assertEquals("2025-01-18 01:02:03.654321+00:00",
IcebergSchemaUtils.serializeInitialDefault(
+ Types.TimestampType.withZone(), 1_737_162_123_654_321L,
flag));
+ }
+ }
+
@Test
public void fieldsWithoutInitialDefaultOmitTheCarrier() {
// A field with no initial default must NOT set initial_default_value
(BE then backfills NULL, the legacy
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTimestampMappingTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTimestampMappingTest.java
index e1ca9b1521d..84ae712d36e 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTimestampMappingTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTimestampMappingTest.java
@@ -43,17 +43,26 @@ class IcebergTimestampMappingTest {
org.apache.iceberg.types.Types.NestedField.optional(1, "ts",
org.apache.iceberg.types.Types.TimestampType.withZone()));
ConnectorType type = ConnectorType.of("TIMESTAMPTZ", 6, 0);
- java.time.LocalDateTime utc = java.time.LocalDateTime.of(1969, 12, 31,
23, 59, 59, 999999000);
- org.apache.doris.connector.spi.pushdown.ConnectorComparison filter =
- new
org.apache.doris.connector.spi.pushdown.ConnectorComparison(
-
org.apache.doris.connector.spi.pushdown.ConnectorComparison.Operator.GE,
- new
org.apache.doris.connector.spi.pushdown.ConnectorColumnRef("ts", type),
- new
org.apache.doris.connector.spi.pushdown.ConnectorLiteral(type, utc));
- for (IcebergPredicateConverter.Mode mode :
IcebergPredicateConverter.Mode.values()) {
- org.apache.iceberg.expressions.Expression expression = new
IcebergPredicateConverter(
- schema, java.time.ZoneId.of("Asia/Shanghai"),
mode).convert(filter).get(0);
- Assertions.assertEquals(-1L,
- ((org.apache.iceberg.expressions.UnboundPredicate<?>)
expression).literal().value());
+ // TIMESTAMPTZ literals already carry UTC fields, including both
instants of a New York DST overlap.
+ for (java.time.LocalDateTime utc : java.util.Arrays.asList(
+ java.time.LocalDateTime.of(1969, 12, 31, 23, 59, 59,
999999000),
+ java.time.LocalDateTime.of(2021, 11, 7, 5, 30, 0, 123456000),
+ java.time.LocalDateTime.of(2021, 11, 7, 6, 30, 0, 123456000)))
{
+ long micros = utc.toEpochSecond(java.time.ZoneOffset.UTC) *
1_000_000L + utc.getNano() / 1000;
+ org.apache.doris.connector.spi.pushdown.ConnectorComparison filter
=
+ new
org.apache.doris.connector.spi.pushdown.ConnectorComparison(
+
org.apache.doris.connector.spi.pushdown.ConnectorComparison.Operator.GE,
+ new
org.apache.doris.connector.spi.pushdown.ConnectorColumnRef("ts", type),
+ new
org.apache.doris.connector.spi.pushdown.ConnectorLiteral(type, utc));
+ for (String zone : java.util.Arrays.asList("UTC", "Asia/Shanghai",
"America/New_York")) {
+ for (IcebergPredicateConverter.Mode mode :
IcebergPredicateConverter.Mode.values()) {
+ org.apache.iceberg.expressions.Expression expression = new
IcebergPredicateConverter(
+ schema, java.time.ZoneId.of(zone),
mode).convert(filter).get(0);
+ Assertions.assertEquals(micros,
+
((org.apache.iceberg.expressions.UnboundPredicate<?>)
expression).literal().value(),
+ utc + " " + zone + " " + mode);
+ }
+ }
}
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
index d917e809378..37e5734bf6f 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
@@ -366,6 +366,19 @@ public class IcebergWritePlanProviderTest {
IcebergWriteSchemaContext.toDorisSql(nestedType,
nestedDefault, false, false));
}
+ @Test
+ public void zonedAndNestedBinaryWriteDefaultsIgnoreLegacyMappingFlags() {
+ // A zoned default keeps its instant, and a nested binary leaf stays
bytes, with the old flags off.
+ Assertions.assertEquals("'2025-01-18 01:02:03.654321+00:00'",
IcebergWriteSchemaContext.toDorisSql(
+ Types.TimestampType.withZone(), 1_737_162_123_654_321L, false,
false));
+ Types.StructType structType = Types.StructType.of(
+ Types.NestedField.required(101, "payload",
Types.BinaryType.get()));
+ GenericRecord structDefault = GenericRecord.create(structType);
+ structDefault.setField("payload", ByteBuffer.wrap(new byte[] {(byte)
0x80, 0x00, (byte) 0xff}));
+ Assertions.assertEquals("named_struct('payload', X'8000FF')",
+ IcebergWriteSchemaContext.toDorisSql(structType,
structDefault, false, false));
+ }
+
@Test
public void branchWritePinsBranchHeadSchema() {
InMemoryCatalog catalog = freshCatalog();
@@ -997,6 +1010,41 @@ public class IcebergWritePlanProviderTest {
Assertions.assertEquals("null",
sink.getStaticPartitionValues().get("id"));
}
+ @Test
+ public void
planWriteKeepsNullBinaryKeysDistinctFromBytesForFullAndHybridSpecs() {
+ String bytes = "0x00112233445566778899aabbccddeeff";
+ for (org.apache.iceberg.types.Type keyType : Arrays.asList(
+ Types.BinaryType.get(), Types.FixedType.ofLength(16),
Types.UUIDType.get())) {
+ for (boolean hybrid : new boolean[] {false, true}) {
+ Schema schema = new Schema(Types.NestedField.optional(1,
"key", keyType),
+ Types.NestedField.optional(2, "region",
Types.StringType.get()));
+ PartitionSpec.Builder spec =
PartitionSpec.builderFor(schema).identity("key");
+ if (hybrid) {
+ spec.identity("region");
+ }
+ Table table =
freshCatalog().createTable(TableIdentifier.of("db1", "t1"), schema,
spec.build(),
+ Collections.singletonMap("write.format.default",
"parquet"));
+ String message = keyType + " hybrid=" + hybrid;
+
+ Map<String, String> nullValues = new
HashMap<>(Collections.singletonMap("key", "NULL"));
+ WriteHandle nullHandle = new WriteHandle(new
IcebergTableHandle("db1", "t1"))
+ .overwrite(true).writeContext(nullValues);
+ nullHandle.staticPartitionNullKeys =
Collections.singleton("key");
+ TIcebergTableSink nullSink = planSink(table,
contextWithStorage(), nullHandle);
+ Assertions.assertEquals(Collections.singleton("key"),
nullSink.getStaticPartitionNullKeys(), message);
+ Assertions.assertEquals("null",
nullSink.getStaticPartitionValues().get("key"), message);
+ // Encoding the wire marker must not rewrite the engine's
static partition spec.
+ Assertions.assertEquals("NULL", nullValues.get("key"),
message);
+
+ TIcebergTableSink bytesSink = planSink(table,
contextWithStorage(),
+ new WriteHandle(new IcebergTableHandle("db1", "t1"))
+
.overwrite(true).writeContext(Collections.singletonMap("key", bytes)));
+
Assertions.assertTrue(bytesSink.getStaticPartitionNullKeys().isEmpty(),
message);
+ Assertions.assertEquals(bytes,
bytesSink.getStaticPartitionValues().get("key"), message);
+ }
+ }
+ }
+
@Test
public void planWriteThreadsBranchFromHandleToTransaction() {
// WHY: INSERT INTO t@branch threads the target branch onto the write
handle; planWrite must hand it
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriterHelperTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriterHelperTest.java
index 441ca3d5bc9..7a7c61de093 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriterHelperTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriterHelperTest.java
@@ -210,6 +210,38 @@ public class IcebergWriterHelperTest {
Assertions.assertEquals(FileFormat.PARQUET, df.format());
}
+ @Test
+ public void convertToWriterResultDecodesBinaryAndUuidPartitionValues() {
+ // BE commits binary-like identity partitions as 0x hex; UUID may also
arrive as canonical text.
+ Schema binarySchema = new Schema(
+ Types.NestedField.optional(1, "key", Types.BinaryType.get()),
+ Types.NestedField.optional(2, "fixed_key",
Types.FixedType.ofLength(2)),
+ Types.NestedField.optional(3, "uuid_key",
Types.UUIDType.get()));
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("test", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db1"));
+ Table table = catalog.createTable(TableIdentifier.of("db1",
"binary_t"), binarySchema,
+
PartitionSpec.builderFor(binarySchema).identity("key").identity("fixed_key").identity("uuid_key")
+ .build(),
+ Collections.singletonMap("write.format.default", "parquet"));
+ java.util.UUID uuid =
java.util.UUID.fromString("00112233-4455-6677-8899-aabbccddeeff");
+ for (String uuidValue :
Arrays.asList("0x00112233445566778899aabbccddeeff", uuid.toString())) {
+ TIcebergCommitData commit = new TIcebergCommitData();
+ commit.setFilePath("s3://b/db1/binary_t/binary.parquet");
+ commit.setPartitionValues(Arrays.asList("0xdead", "0xdead",
uuidValue));
+ commit.setRowCount(1L);
+ commit.setFileSize(128L);
+ commit.setFileContent(TFileContent.DATA);
+ DataFile file = IcebergWriterHelper.convertToWriterResult(
+ table, Collections.singletonList(commit),
ZoneOffset.UTC).dataFiles()[0];
+ Assertions.assertEquals(ByteBuffer.wrap(new byte[] {(byte) 0xde,
(byte) 0xad}),
+ file.partition().get(0, ByteBuffer.class));
+ Assertions.assertEquals(file.partition().get(0, ByteBuffer.class),
+ file.partition().get(1, ByteBuffer.class));
+ Assertions.assertEquals(uuid, file.partition().get(2,
java.util.UUID.class), uuidValue);
+ }
+ }
+
@Test
public void convertToWriterResultCarriesColumnMetrics() {
Table table = tableWith("write.format.default", "parquet");
diff --git
a/fe/fe-connector/fe-connector-jdbc/src/test/java/org/apache/doris/connector/jdbc/JdbcTimestampProjectionTest.java
b/fe/fe-connector/fe-connector-jdbc/src/test/java/org/apache/doris/connector/jdbc/JdbcTimestampProjectionTest.java
index 27d267f0e2c..b6c8c553da1 100644
---
a/fe/fe-connector/fe-connector-jdbc/src/test/java/org/apache/doris/connector/jdbc/JdbcTimestampProjectionTest.java
+++
b/fe/fe-connector/fe-connector-jdbc/src/test/java/org/apache/doris/connector/jdbc/JdbcTimestampProjectionTest.java
@@ -18,10 +18,25 @@
package org.apache.doris.connector.jdbc;
import org.apache.doris.connector.spi.ConnectorType;
+import org.apache.doris.connector.spi.handle.ConnectorColumnHandle;
+import org.apache.doris.connector.spi.pushdown.ConnectorColumnRef;
+import org.apache.doris.connector.spi.pushdown.ConnectorComparison;
+import org.apache.doris.connector.spi.pushdown.ConnectorExpression;
+import org.apache.doris.connector.spi.pushdown.ConnectorFunctionCall;
+import org.apache.doris.connector.spi.pushdown.ConnectorIn;
+import org.apache.doris.connector.spi.pushdown.ConnectorIsNull;
+import org.apache.doris.connector.spi.pushdown.ConnectorLiteral;
+import org.apache.doris.connector.spi.pushdown.ConnectorOr;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
class JdbcTimestampProjectionTest {
@Test
void trinoSessionClauseRemainsOutsideTimestampProjection() {
@@ -201,4 +216,100 @@ class JdbcTimestampProjectionTest {
}
}
+ private static final ConnectorType INSTANT =
ConnectorType.of("TIMESTAMPTZ", 6, 0);
+ private static final ConnectorType LOCAL = ConnectorType.of("DATETIMEV2",
6, 0);
+ private static final ConnectorType BYTES = ConnectorType.of("VARBINARY",
16, 0);
+ private static final List<ConnectorColumnHandle> FILTER_COLUMNS =
Arrays.asList(
+ new JdbcColumnHandle("ts", "ts", INSTANT), new
JdbcColumnHandle("other_ts", "other_ts", INSTANT),
+ new JdbcColumnHandle("dt", "dt", LOCAL), new
JdbcColumnHandle("payload", "payload", BYTES),
+ new JdbcColumnHandle("id", "id", ConnectorType.of("INT")));
+
+ private static boolean pushedDown(JdbcQueryBuilder builder,
ConnectorExpression filter) {
+ return builder.buildQuery("db", "tbl", FILTER_COLUMNS,
Optional.of(filter), -1).contains("WHERE");
+ }
+
+ private static ConnectorExpression idIsOne() {
+ return new ConnectorComparison(ConnectorComparison.Operator.EQ,
+ new ConnectorColumnRef("id", ConnectorType.of("INT")), new
ConnectorLiteral(ConnectorType.of("INT"), 1));
+ }
+
+ @Test
+ void calendarFunctionsOnInstantsStayLocal() {
+ JdbcQueryBuilder mysql = new JdbcQueryBuilder(JdbcDbType.MYSQL,
+ JdbcFunctionPushdownConfig.create(JdbcDbType.MYSQL, null,
true), Collections.emptyMap());
+ // year() of an instant depends on the zone; the remote UTC session
differs from the Doris session.
+ Assertions.assertFalse(pushedDown(mysql, new
ConnectorComparison(ConnectorComparison.Operator.EQ,
+ new ConnectorFunctionCall("year", ConnectorType.of("INT"),
+ Collections.singletonList(new ConnectorColumnRef("ts",
INSTANT))),
+ new ConnectorLiteral(ConnectorType.of("INT"), 2020))));
+ Assertions.assertTrue(pushedDown(mysql, new
ConnectorComparison(ConnectorComparison.Operator.EQ,
+ new ConnectorFunctionCall("year", ConnectorType.of("INT"),
+ Collections.singletonList(new ConnectorColumnRef("dt",
LOCAL))),
+ new ConnectorLiteral(ConnectorType.of("INT"), 2020))));
+ // A direct null check is zone-independent.
+ Assertions.assertTrue(pushedDown(mysql, new ConnectorIsNull(new
ConnectorColumnRef("ts", INSTANT), false)));
+ }
+
+ @Test
+ void instantLiteralsStayLocalInEveryDialect() {
+ ConnectorLiteral literal = new ConnectorLiteral(INSTANT,
LocalDateTime.of(2020, 1, 2, 4, 1, 0, 111333000));
+ for (JdbcDbType dialect : JdbcDbType.values()) {
+ JdbcQueryBuilder builder = new JdbcQueryBuilder(dialect);
+ Assertions.assertFalse(pushedDown(builder, new
ConnectorComparison(ConnectorComparison.Operator.EQ,
+ new ConnectorColumnRef("ts", INSTANT), literal)),
dialect.name());
+ Assertions.assertFalse(pushedDown(builder, new ConnectorIn(new
ConnectorColumnRef("ts", INSTANT),
+ Collections.singletonList(literal), false)),
dialect.name());
+ }
+ }
+
+ @Test
+ void postgresInstantPredicatesObserveDecodedNulls() {
+ // Out-of-range PostgreSQL instants become NULL only after JDBC
decoding.
+ JdbcQueryBuilder postgres = new
JdbcQueryBuilder(JdbcDbType.POSTGRESQL);
+ ConnectorColumnRef ts = new ConnectorColumnRef("ts", INSTANT);
+ ConnectorColumnRef other = new ConnectorColumnRef("other_ts", INSTANT);
+ for (ConnectorExpression predicate : Arrays.asList(new
ConnectorIsNull(ts, false),
+ new ConnectorIsNull(ts, true),
+ new ConnectorComparison(ConnectorComparison.Operator.EQ, ts,
other),
+ new
ConnectorComparison(ConnectorComparison.Operator.EQ_FOR_NULL, ts, other),
+ new ConnectorOr(Arrays.asList(new ConnectorIsNull(ts, false),
idIsOne())))) {
+ Assertions.assertFalse(pushedDown(postgres, predicate),
predicate.toString());
+ }
+ Assertions.assertTrue(pushedDown(new
JdbcQueryBuilder(JdbcDbType.MYSQL), new ConnectorIsNull(ts, false)));
+ Assertions.assertTrue(pushedDown(postgres, new ConnectorIsNull(new
ConnectorColumnRef("dt", LOCAL), false)));
+ Assertions.assertTrue(pushedDown(postgres, idIsOne()));
+ }
+
+ @Test
+ void binaryLiteralsStayLocalWhereTheDialectReadsThemDifferently() {
+ ConnectorColumnRef payload = new ConnectorColumnRef("payload", BYTES);
+ for (JdbcDbType dialect : Arrays.asList(JdbcDbType.POSTGRESQL,
JdbcDbType.ORACLE, JdbcDbType.SQLSERVER,
+ JdbcDbType.DB2, JdbcDbType.OCEANBASE_ORACLE)) {
+ JdbcQueryBuilder builder = new JdbcQueryBuilder(dialect);
+ for (ConnectorComparison.Operator operator :
Arrays.asList(ConnectorComparison.Operator.EQ,
+ ConnectorComparison.Operator.NE,
ConnectorComparison.Operator.LT)) {
+ Assertions.assertFalse(pushedDown(builder, new
ConnectorComparison(operator, payload,
+ binaryLiteral((byte) 0, (byte) 0xff))), dialect + " "
+ operator);
+ }
+ for (boolean notIn : new boolean[] {false, true}) {
+ ConnectorExpression in = new ConnectorIn(payload,
Arrays.asList(
+ binaryLiteral((byte) 0xde, (byte) 0, (byte) 0xff),
binaryLiteral()), notIn);
+ Assertions.assertFalse(pushedDown(builder, in), dialect + "
notIn=" + notIn);
+ Assertions.assertFalse(pushedDown(builder, new
ConnectorOr(Arrays.asList(in, idIsOne()))),
+ dialect + " notIn=" + notIn);
+ }
+ Assertions.assertTrue(pushedDown(builder, new ConnectorIn(new
ConnectorColumnRef("id",
+ ConnectorType.of("INT")), Collections.singletonList(
+ new ConnectorLiteral(ConnectorType.of("INT"), 1)),
false)), dialect.name());
+ }
+ String mysql = new JdbcQueryBuilder(JdbcDbType.MYSQL).buildQuery("db",
"tbl", FILTER_COLUMNS, Optional.of(
+ new ConnectorIn(payload,
Collections.singletonList(binaryLiteral((byte) 0, (byte) 0xff)), false)), -1);
+ Assertions.assertTrue(mysql.contains("IN (X'00ff')"), mysql);
+ }
+
+ // VARBINARY literals cross the SPI as a lossless Latin-1 string carrier.
+ private static ConnectorLiteral binaryLiteral(byte... bytes) {
+ return new ConnectorLiteral(BYTES, new String(bytes,
java.nio.charset.StandardCharsets.ISO_8859_1));
+ }
+
}
diff --git
a/fe/fe-connector/fe-connector-jdbc/src/test/java/org/apache/doris/connector/jdbc/client/JdbcLogicalTypeMappingTest.java
b/fe/fe-connector/fe-connector-jdbc/src/test/java/org/apache/doris/connector/jdbc/client/JdbcLogicalTypeMappingTest.java
new file mode 100644
index 00000000000..a156d2f0002
--- /dev/null
+++
b/fe/fe-connector/fe-connector-jdbc/src/test/java/org/apache/doris/connector/jdbc/client/JdbcLogicalTypeMappingTest.java
@@ -0,0 +1,120 @@
+// 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.doris.connector.jdbc.client;
+
+import org.apache.doris.connector.jdbc.JdbcDbType;
+import org.apache.doris.connector.spi.ConnectorType;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Types;
+import java.util.Collections;
+import java.util.Optional;
+
+/**
+ * The binary and instant mappings are determined by source semantics; the
deprecated
+ * {@code enable.mapping.*} flags must not select a different external logical
type.
+ */
+class JdbcLogicalTypeMappingTest {
+
+ @Test
+ void binaryTypesKeepBytesAndDeclaredLength() {
+ for (boolean flag : new boolean[] {false, true}) {
+ assertVarbinary(new JdbcMySQLConnectorClient("test",
JdbcDbType.MYSQL, "jdbc:mysql://localhost/test",
+ false, Collections.emptyMap(), Collections.emptyMap(),
flag, flag),
+ "BINARY", "VARBINARY", "TINYBLOB", "BLOB", "MEDIUMBLOB",
"LONGBLOB");
+ assertVarbinary(postgres(flag), "bytea");
+ assertVarbinary(new JdbcOracleConnectorClient("test",
JdbcDbType.ORACLE, "jdbc:oracle:thin:@localhost",
+ false, Collections.emptyMap(), Collections.emptyMap(),
flag, flag), "BLOB");
+ assertVarbinary(sqlServer(flag), "binary", "varbinary", "image");
+ assertVarbinary(new JdbcDB2ConnectorClient("test", JdbcDbType.DB2,
"jdbc:db2://localhost/test",
+ false, Collections.emptyMap(), Collections.emptyMap(),
flag, flag), "BLOB", "BINARY", "VARBINARY");
+ }
+ }
+
+ @Test
+ void instantTimestampsMapToTimestamptzWhileLocalTimestampsStayDatetime() {
+ for (boolean flag : new boolean[] {false, true}) {
+ JdbcConnectorClient mysql = new JdbcMySQLConnectorClient("test",
JdbcDbType.MYSQL,
+ "jdbc:mysql://localhost/test", false,
Collections.emptyMap(), Collections.emptyMap(), flag, flag);
+ assertTimestamp(mysql, "TIMESTAMP", true);
+ assertTimestamp(mysql, "DATETIME", false);
+ assertTimestamp(postgres(flag), "timestamptz", true);
+ assertTimestamp(postgres(flag), "timestamp", false);
+ JdbcConnectorClient oracle = new JdbcOracleConnectorClient("test",
JdbcDbType.ORACLE,
+ "jdbc:oracle:thin:@localhost", false,
Collections.emptyMap(), Collections.emptyMap(), flag, flag);
+ assertTimestamp(oracle, "TIMESTAMP(6) WITH LOCAL TIME ZONE", true);
+ assertTimestamp(oracle, "TIMESTAMP(6) WITH TIME ZONE", true);
+ assertTimestamp(oracle, "TIMESTAMP(6)", false);
+ JdbcConnectorClient trino = new JdbcTrinoConnectorClient("test",
JdbcDbType.TRINO,
+ "jdbc:trino://localhost", false, Collections.emptyMap(),
Collections.emptyMap(), flag, flag);
+ assertTimestamp(trino, "timestamp(6) with time zone", true);
+ assertTimestamp(trino, "timestamp(6)", false);
+ assertTimestamp(sqlServer(flag), "datetimeoffset", true);
+ assertTimestamp(sqlServer(flag), "datetime2", false);
+ assertTimestamp(new JdbcClickHouseConnectorClient("test",
JdbcDbType.CLICKHOUSE,
+ "jdbc:clickhouse://localhost/test", false,
Collections.emptyMap(), Collections.emptyMap(),
+ flag, flag), "DateTime64(6, 'UTC')", true);
+ }
+ }
+
+ @Test
+ void postgresInstantsAllowDecoderCreatedNulls() {
+ // Out-of-range and infinite timestamptz values decode to NULL, so a
source NOT NULL cannot hold.
+ JdbcFieldInfo instant = timestampField("timestamptz");
+ instant.setAllowNull(false);
+ postgres(false).jdbcTypeToConnectorType(instant);
+ Assertions.assertTrue(instant.isAllowNull());
+
+ JdbcFieldInfo local = timestampField("timestamp");
+ local.setAllowNull(false);
+ postgres(false).jdbcTypeToConnectorType(local);
+ Assertions.assertFalse(local.isAllowNull());
+ }
+
+ private static JdbcConnectorClient postgres(boolean flag) {
+ return new JdbcPostgreSQLConnectorClient("test",
JdbcDbType.POSTGRESQL, "jdbc:postgresql://localhost/test",
+ false, Collections.emptyMap(), Collections.emptyMap(), flag,
flag);
+ }
+
+ private static JdbcConnectorClient sqlServer(boolean flag) {
+ return new JdbcSQLServerConnectorClient("test", JdbcDbType.SQLSERVER,
"jdbc:sqlserver://localhost",
+ false, Collections.emptyMap(), Collections.emptyMap(), flag,
flag);
+ }
+
+ private static JdbcFieldInfo timestampField(String typeName) {
+ return new JdbcFieldInfo("ts", Optional.of(typeName), Types.TIMESTAMP,
+ Optional.of(26), Optional.of(6), Optional.empty());
+ }
+
+ private static void assertVarbinary(JdbcConnectorClient client, String...
typeNames) {
+ for (String typeName : typeNames) {
+ ConnectorType type = client.jdbcTypeToConnectorType(new
JdbcFieldInfo("b", Optional.of(typeName),
+ Types.VARBINARY, Optional.of(32), Optional.of(0),
Optional.empty()));
+ Assertions.assertEquals("VARBINARY", type.getTypeName(), typeName);
+ Assertions.assertEquals(32, type.getPrecision(), typeName);
+ }
+ }
+
+ private static void assertTimestamp(JdbcConnectorClient client, String
typeName, boolean instant) {
+ ConnectorType type =
client.jdbcTypeToConnectorType(timestampField(typeName));
+ Assertions.assertEquals(instant ? "TIMESTAMPTZ" : "DATETIMEV2",
type.getTypeName(), typeName);
+ Assertions.assertEquals(6, type.getPrecision(), typeName);
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonLegacyOrcTimestampTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonLegacyOrcTimestampTest.java
index a0840317cc3..7d0221494b8 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonLegacyOrcTimestampTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonLegacyOrcTimestampTest.java
@@ -33,11 +33,14 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
class PaimonLegacyOrcTimestampTest {
@Test
@@ -75,4 +78,46 @@ class PaimonLegacyOrcTimestampTest {
Collections.singleton(1), new HashMap<>()));
}
}
+
+ @Test
+ void routeEveryNestedLtzShapeAndMixedSplitToSdk(@TempDir Path warehouse)
throws Exception {
+ org.apache.paimon.types.DataType ltz =
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(6);
+ List<org.apache.paimon.types.DataType> shapes = Arrays.asList(ltz,
DataTypes.ARRAY(ltz),
+ DataTypes.MULTISET(ltz), DataTypes.MAP(ltz, DataTypes.INT()),
DataTypes.MAP(DataTypes.INT(), ltz),
+ DataTypes.ROW(DataTypes.FIELD(1, "ts", ltz)));
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ for (int i = 0; i < shapes.size(); i++) {
+ // An absent option means Paimon's legacy default, so both
spellings must route to the SDK.
+ for (boolean explicit : new boolean[] {false, true}) {
+ Identifier id = Identifier.create("db", "t" + i +
(explicit ? "_explicit" : "_default"));
+ Schema.Builder schema = Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("c", shapes.get(i));
+ if (explicit) {
+
schema.option(OrcOptions.ORC_TIMESTAMP_LTZ_LEGACY_TYPE.key(), "true");
+ }
+ catalog.createTable(id, schema.build(), false);
+ FileStoreTable table = (FileStoreTable)
catalog.getTable(id);
+ Set<Integer> readC =
Collections.singleton(table.rowType().getField("c").id());
+ String message = shapes.get(i) + (explicit ? " explicit" :
" default");
+ Assertions.assertTrue(requiresSdk(table, readC,
"data.orc"), message);
+ // One legacy ORC file is enough to route a mixed split
away from the native reader.
+ Assertions.assertTrue(requiresSdk(table, readC,
"data.parquet", "data.orc"), message);
+ Assertions.assertFalse(requiresSdk(table, readC,
"data.parquet"), message);
+ }
+ }
+ }
+ }
+
+ private static boolean requiresSdk(FileStoreTable table, Set<Integer>
readFieldIds, String... paths) {
+ List<RawFile> files = new ArrayList<>();
+ for (String path : paths) {
+ files.add(new RawFile(path, 0L, 100L, 1L,
path.substring(path.lastIndexOf('.') + 1),
+ table.schema().id(), 0L));
+ }
+ return PaimonScanPlanProvider.requiresLegacyOrcTimestampReader(
+ table, Optional.of(files), readFieldIds, new HashMap<>());
+ }
}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPartitionValueRenderTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPartitionValueRenderTest.java
index 0da0c76c093..1af43861c8c 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPartitionValueRenderTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPartitionValueRenderTest.java
@@ -24,6 +24,9 @@ import org.junit.jupiter.api.Test;
import java.time.LocalDate;
import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.Arrays;
/**
* Unit tests for {@link PaimonScanPlanProvider#serializePartitionValue}, the
FIX-NATIVE-PARTVAL
@@ -66,6 +69,22 @@ public class PaimonPartitionValueRenderTest {
Assertions.assertEquals("2024-01-01T01:02:03Z", utc);
}
+ @Test
+ public void ltzRetainsPreEpochAndDstFoldInstantsInEverySessionZone() {
+ // Both New York fold instants display as 01:30; only the explicit
offset keeps them apart.
+ for (LocalDateTime utc : Arrays.asList(
+ LocalDateTime.of(1969, 12, 31, 23, 59, 59, 999999000),
+ LocalDateTime.of(2021, 11, 7, 5, 30, 0, 123456000),
+ LocalDateTime.of(2021, 11, 7, 6, 30, 0, 123456000))) {
+ for (String zone : Arrays.asList("UTC", "Asia/Shanghai",
"America/New_York")) {
+ String value = PaimonScanPlanProvider.serializePartitionValue(
+ DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(6),
Timestamp.fromLocalDateTime(utc), zone);
+ Assertions.assertEquals(utc.toInstant(ZoneOffset.UTC),
+ OffsetDateTime.parse(value).toInstant(), utc + " " +
zone);
+ }
+ }
+ }
+
@Test
public void ntzRendersIsoNoZoneShift() {
Timestamp ts = Timestamp.fromLocalDateTime(LocalDateTime.of(2024, 1,
1, 1, 2, 3));
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
index 70fe9cc4993..9223305efbb 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
@@ -2791,7 +2791,7 @@ public class PaimonScanPlanProviderTest {
.column("id", DataTypes.INT().notNull())
.column("event_time", new
org.apache.paimon.types.LocalZonedTimestampType(6))
.primaryKey("id").option("bucket", "1")
- .option("file.format",
"orc").option("orc.timestamp-ltz.legacy-type", "true")
+ .option("file.format",
"orc").option("orc.timestamp-ltz.legacy.type", "true")
.build(), false);
Table table = catalog.getTable(id);
BatchWriteBuilder builder = table.newBatchWriteBuilder();
@@ -2857,7 +2857,7 @@ public class PaimonScanPlanProviderTest {
catalog.createTable(id, Schema.newBuilder()
.column("id", DataTypes.INT().notNull())
.primaryKey("id").option("bucket", "1")
- .option("file.format",
"orc").option("orc.timestamp-ltz.legacy-type", "true")
+ .option("file.format",
"orc").option("orc.timestamp-ltz.legacy.type", "true")
.build(), false);
Table table = catalog.getTable(id);
BatchWriteBuilder builder = table.newBatchWriteBuilder();
@@ -2972,7 +2972,7 @@ public class PaimonScanPlanProviderTest {
.column("id", DataTypes.INT())
.column("event_time", new
org.apache.paimon.types.LocalZonedTimestampType(6))
.option("file.format", "orc")
- .option("orc.timestamp-ltz.legacy-type", "true")
+ .option("orc.timestamp-ltz.legacy.type", "true")
.build(), false);
Table table = catalog.getTable(id);
BatchWriteBuilder builder = table.newBatchWriteBuilder();
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWritePlanProviderTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWritePlanProviderTest.java
index 8995f4fc5c0..cd321088f72 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWritePlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWritePlanProviderTest.java
@@ -110,6 +110,27 @@ public class PaimonWritePlanProviderTest {
2, 3, false, CoreOptions.MergeEngine.FIRST_ROW));
}
+ @Test
+ public void
writeColumnsKeepInstantSemanticsInsideContainersRegardlessOfLegacyFlags() {
+ // A civil DATETIMEV2 write boundary would collapse both instants of a
DST fold.
+ List<DataField> fields = Arrays.asList(
+ DataTypes.FIELD(0, "instant",
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(6)),
+ DataTypes.FIELD(1, "local_time", DataTypes.TIMESTAMP(6)),
+ DataTypes.FIELD(2, "items",
DataTypes.ARRAY(DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(3))),
+ DataTypes.FIELD(3, "record", DataTypes.ROW(
+ DataTypes.FIELD(4, "instant",
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(6)))));
+ for (boolean flag : new boolean[] {false, true}) {
+ List<ConnectorColumn> columns =
PaimonWritePlanProvider.mapWriteColumns(
+ fields, Collections.emptyList(), new
PaimonTypeMapping.Options(flag, flag));
+ Assertions.assertEquals(ConnectorType.of("TIMESTAMPTZ", 6, 0),
columns.get(0).getType());
+ Assertions.assertEquals(ConnectorType.of("DATETIMEV2", 6, 0),
columns.get(1).getType());
+ Assertions.assertEquals(ConnectorType.of("TIMESTAMPTZ", 3, 0),
+ columns.get(2).getType().getChildren().get(0));
+ Assertions.assertEquals(ConnectorType.of("TIMESTAMPTZ", 6, 0),
+ columns.get(3).getType().getChildren().get(0));
+ }
+ }
+
@Test
public void
writeColumnsConvertPaimonDefaultsToDorisSqlAndPreserveSchemaProperties() {
List<ConnectorColumn> columns =
PaimonWritePlanProvider.mapWriteColumns(
diff --git
a/fe/fe-connector/fe-connector-trino/src/test/java/org/apache/doris/connector/trino/TrinoTimestampMappingTest.java
b/fe/fe-connector/fe-connector-trino/src/test/java/org/apache/doris/connector/trino/TrinoTimestampMappingTest.java
index 616215e6257..6e05570e11c 100644
---
a/fe/fe-connector/fe-connector-trino/src/test/java/org/apache/doris/connector/trino/TrinoTimestampMappingTest.java
+++
b/fe/fe-connector/fe-connector-trino/src/test/java/org/apache/doris/connector/trino/TrinoTimestampMappingTest.java
@@ -62,6 +62,46 @@ class TrinoTimestampMappingTest {
}
}
+ @Test
+ void domainsEncodeUtcMicrosAndMillisecondAlignedZonedBounds() {
+ for (java.time.LocalDateTime utc : java.util.Arrays.asList(
+ java.time.LocalDateTime.of(2020, 1, 2, 4, 1, 0, 111333000),
+ java.time.LocalDateTime.of(1969, 12, 31, 23, 59, 59,
999999000))) {
+ long millis =
utc.toInstant(java.time.ZoneOffset.UTC).toEpochMilli();
+ // TIMESTAMPTZ(6) keeps the sub-millisecond part as picoseconds of
the UTC millisecond.
+
Assertions.assertEquals(io.trino.spi.type.LongTimestampWithTimeZone.fromEpochMillisAndFraction(
+ millis, (utc.getNano() % 1_000_000) * 1000,
io.trino.spi.type.TimeZoneKey.UTC_KEY),
+
singleDomainValue(io.trino.spi.type.TimestampWithTimeZoneType.createTimestampWithTimeZoneType(6),
+ "TIMESTAMPTZ", 6, utc));
+ // TIMESTAMPTZ(3) accepts a millisecond-aligned bound as a packed
UTC value.
+ java.time.LocalDateTime aligned = utc.withNano(utc.getNano() /
1_000_000 * 1_000_000);
+ long packed = (long) singleDomainValue(
+
io.trino.spi.type.TimestampWithTimeZoneType.createTimestampWithTimeZoneType(3),
+ "TIMESTAMPTZ", 3, aligned);
+
Assertions.assertEquals(aligned.toInstant(java.time.ZoneOffset.UTC).toEpochMilli(),
+
io.trino.spi.type.DateTimeEncoding.unpackMillisUtc(packed));
+ // TIMESTAMP(6) carries the wall-clock fields as epoch
microseconds without double counting millis.
+
Assertions.assertEquals(utc.toEpochSecond(java.time.ZoneOffset.UTC) *
1_000_000L + utc.getNano() / 1000,
+
singleDomainValue(io.trino.spi.type.TimestampType.createTimestampType(6),
"DATETIMEV2", 6, utc));
+ }
+ }
+
+ private static Object singleDomainValue(io.trino.spi.type.Type trinoType,
String dorisTypeName, int precision,
+ java.time.LocalDateTime value) {
+ io.trino.spi.connector.ColumnHandle handle = new
io.trino.spi.connector.ColumnHandle() { };
+ org.apache.doris.connector.spi.ConnectorType dorisType =
+ org.apache.doris.connector.spi.ConnectorType.of(dorisTypeName,
precision, 0);
+ org.apache.doris.connector.spi.pushdown.ConnectorExpression filter =
+ new
org.apache.doris.connector.spi.pushdown.ConnectorComparison(
+
org.apache.doris.connector.spi.pushdown.ConnectorComparison.Operator.EQ,
+ new
org.apache.doris.connector.spi.pushdown.ConnectorColumnRef("ts", dorisType),
+ new
org.apache.doris.connector.spi.pushdown.ConnectorLiteral(dorisType, value));
+ TrinoPredicateConverter converter = new TrinoPredicateConverter(
+ java.util.Collections.singletonMap("ts", handle),
+ java.util.Collections.singletonMap("ts", new
io.trino.spi.connector.ColumnMetadata("ts", trinoType)));
+ return
converter.convert(filter).getDomains().orElseThrow(AssertionError::new).get(handle).getSingleValue();
+ }
+
@Test
void subMicrosecondLocalTimestampDomainStaysLocal() {
io.trino.spi.connector.ColumnHandle handle = new
io.trino.spi.connector.ColumnHandle() { };
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/jdbc/client/JdbcClientTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/jdbc/client/JdbcClientTest.java
new file mode 100644
index 00000000000..30f11fee297
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/jdbc/client/JdbcClientTest.java
@@ -0,0 +1,127 @@
+// 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.doris.datasource.jdbc.client;
+
+import org.apache.doris.catalog.ArrayType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.datasource.jdbc.util.JdbcFieldSchema;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.sql.ResultSet;
+import java.sql.Types;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * The legacy fe-core clients (still used by CDC schema discovery) map binary
and instant types by source
+ * semantics; the deprecated mapping flags must not restore the STRING or
unzoned DATETIME contract.
+ */
+public class JdbcClientTest {
+
+ @Test
+ public void testBinaryMappingsKeepBytesAndDeclaredLength() throws
Exception {
+ assertBinaryMapping(JdbcMySQLClient.class, "BINARY", "VARBINARY",
"TINYBLOB", "BLOB", "MEDIUMBLOB", "LONGBLOB");
+ assertBinaryMapping(JdbcPostgreSQLClient.class, "bytea");
+ assertBinaryMapping(JdbcOracleClient.class, "BLOB");
+ assertBinaryMapping(JdbcSQLServerClient.class, "binary", "varbinary",
"image");
+ assertBinaryMapping(JdbcDB2Client.class, "BLOB");
+ JdbcClientConfig config = new
JdbcClientConfig().setEnableMappingVarbinary(false);
+ Assertions.assertTrue(config.isEnableMappingVarbinary());
+ }
+
+ private void assertBinaryMapping(Class<? extends JdbcClient> clientClass,
String... names) throws Exception {
+ JdbcClient client = Mockito.mock(clientClass,
Mockito.CALLS_REAL_METHODS);
+ for (String name : names) {
+ JdbcFieldSchema field = new
JdbcFieldSchema(Mockito.mock(ResultSet.class));
+ field.setDataTypeName(Optional.of(name));
+ field.setColumnSize(Optional.of(32));
+ Assertions.assertEquals(ScalarType.createVarbinaryType(32),
client.jdbcTypeToDoris(field), name);
+ }
+ }
+
+ @Test
+ public void testTimestampMappingsIgnoreLegacyFlag() throws Exception {
+ assertTimestampMapping(JdbcMySQLClient.class, "TIMESTAMP", true);
+ assertTimestampMapping(JdbcMySQLClient.class, "DATETIME", false);
+ assertTimestampMapping(JdbcPostgreSQLClient.class, "timestamptz",
true);
+ assertTimestampMapping(JdbcPostgreSQLClient.class, "timestamp", false);
+ assertTimestampMapping(JdbcOracleClient.class, "TIMESTAMP(6) WITH
LOCAL TIME ZONE", true);
+ assertTimestampMapping(JdbcOracleClient.class, "TIMESTAMP(6)", false);
+ }
+
+ @Test
+ public void testExplicitZonedJdbcTimestampMappings() throws Exception {
+ assertTimestampMapping(JdbcTrinoClient.class, "timestamp(6) with time
zone", true);
+ assertTimestampMapping(JdbcTrinoClient.class, "timestamp(6)", false);
+ assertTimestampMapping(JdbcSQLServerClient.class, "datetimeoffset",
true);
+ assertTimestampMapping(JdbcSQLServerClient.class, "datetime2", false);
+ assertTimestampMapping(JdbcOracleClient.class, "TIMESTAMP(6) WITH TIME
ZONE", true);
+ assertTimestampMapping(JdbcClickHouseClient.class, "DateTime64(6,
'UTC')", true);
+ }
+
+ private void assertTimestampMapping(Class<? extends JdbcClient>
clientClass, String typeName,
+ boolean withZone) throws Exception {
+ // Avoid opening remote connections; execute the real metadata
conversion with either legacy flag.
+ JdbcClient client = Mockito.mock(clientClass,
Mockito.CALLS_REAL_METHODS);
+ JdbcFieldSchema field = new
JdbcFieldSchema(Mockito.mock(ResultSet.class));
+ field.setDataTypeName(Optional.of(typeName));
+ field.setColumnSize(Optional.of(26));
+ field.setDecimalDigits(Optional.of(6));
+ for (boolean flag : new boolean[] {false, true}) {
+ client.enableMappingTimestampTz = flag;
+ Assertions.assertEquals(withZone ?
ScalarType.createTimeStampTzType(6)
+ : ScalarType.createDatetimeV2Type(6),
client.jdbcTypeToDoris(field), typeName);
+ }
+ }
+
+ @Test
+ public void testPostgresTimestampMappingAllowsDecoderCreatedNulls() throws
Exception {
+ JdbcPostgreSQLClient client = Mockito.mock(JdbcPostgreSQLClient.class,
Mockito.CALLS_REAL_METHODS);
+ JdbcFieldSchema instant = new
JdbcFieldSchema(Mockito.mock(ResultSet.class));
+ instant.setColumnName("event_time");
+ instant.setDataTypeName(Optional.of("timestamptz"));
+ instant.setDecimalDigits(Optional.of(6));
+ instant.setAllowNull(false);
+ JdbcFieldSchema local = new JdbcFieldSchema(instant);
+ local.setColumnName("local_time");
+ local.setDataTypeName(Optional.of("timestamp"));
+ JdbcFieldSchema array = new JdbcFieldSchema(instant);
+ array.setColumnName("events");
+ array.setDataTypeName(Optional.of("_timestamptz"));
+ array.setDataType(Types.ARRAY);
+ array.setArrayDimensions(Optional.of(2));
+ Mockito.doReturn(Arrays.asList(instant, local, array)).when(client)
+ .getJdbcColumnsInfo("sample_schema", "sample_table");
+
+ List<Column> columns = client.getColumnsFromJdbc("sample_schema",
"sample_table");
+
+ // Source NOT NULL cannot prevent the timestamp range conversion from
creating NULL.
+ Assertions.assertTrue(columns.get(0).isAllowNull());
+ Assertions.assertTrue(columns.get(0).getType().isTimeStampTz());
+ Assertions.assertFalse(columns.get(1).isAllowNull());
+ Assertions.assertFalse(columns.get(2).isAllowNull());
+ ArrayType arrayType = (ArrayType) columns.get(2).getType();
+ Assertions.assertTrue(arrayType.getContainsNull());
+ Assertions.assertTrue(((ArrayType)
arrayType.getItemType()).getContainsNull());
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java
index 4496694c284..f232aa9f0cc 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java
@@ -1443,6 +1443,35 @@ public class PluginDrivenMvccExternalTableTest {
"the NULL-min partition's exclusive upper must be
lowerKey.successor()");
}
+ @Test
+ public void testRangeViewTimestamptzBoundsStayUtcInEverySessionZone()
throws AnalysisException {
+ // Transform bounds are UTC instants, and the NULL-min sentinel must
stay below them: parsing it as a
+ // wall-clock date would underflow year zero in a positive-offset
session.
+ Type timestamptz =
org.apache.doris.catalog.ScalarType.createTimeStampTzType(6);
+ for (String zone : Arrays.asList("UTC", "Asia/Shanghai",
"America/New_York")) {
+ ConnectContext ctx = new ConnectContext();
+ ctx.getSessionVariable().setTimeZone(zone);
+ ctx.setThreadLocalInfo();
+ try {
+ Fixture f = Fixture.rangeView(rangeView(
+ rangePart("pnull", "0000-01-01", null, FRESH_555),
+ rangePart("p0", "1970-01-01 00:00:00+00:00",
"1970-01-02 00:00:00+00:00", FRESH_777)),
+ timestamptz);
+ Map<String, PartitionItem> items =
f.table.getNameToPartitionItems(Optional.empty());
+ com.google.common.collect.Range<PartitionKey> nullRange =
+ ((RangePartitionItem) items.get("pnull")).getItems();
+ com.google.common.collect.Range<PartitionKey> dayZero =
+ ((RangePartitionItem) items.get("p0")).getItems();
+ Assertions.assertFalse(nullRange.isEmpty(), zone);
+
Assertions.assertTrue(nullRange.upperEndpoint().compareTo(dayZero.lowerEndpoint())
< 0, zone);
+ Assertions.assertEquals(0,
dayZero.lowerEndpoint().getKeys().get(0).compareLiteral(
+ new org.apache.doris.analysis.DateLiteral(1970, 1, 1,
0, 0, 0, 0, timestamptz)), zone);
+ } finally {
+ ConnectContext.remove();
+ }
+ }
+ }
+
@Test
public void testRangeViewPartitionSnapshotIsSnapshotId() throws
AnalysisException {
Fixture f = Fixture.rangeView(rangeView(
@@ -1709,7 +1738,13 @@ public class PluginDrivenMvccExternalTableTest {
* the snapshot-consistent enumeration. The partition column is the
default DATEV2 {@code dt}.
*/
static Fixture rangeView(ConnectorMvccPartitionView view) {
- Fixture f = partitioned();
+ return rangeView(view, Type.DATEV2);
+ }
+
+ static Fixture rangeView(ConnectorMvccPartitionView view, Type
partitionColType) {
+ Fixture f = with(Arrays.asList(
+ cpi("dt=2024-01-01", TS_2024_01_01),
+ cpi("dt=2024-02-02", TS_2024_02_02)), partitionColType);
Mockito.when(f.metadata.applySnapshot(Mockito.eq(f.session),
Mockito.eq(f.handle), Mockito.any()))
.thenReturn(f.pinnedHandle);
Mockito.when(f.metadata.getMvccPartitionView(f.session,
f.pinnedHandle))
diff --git
a/regression-test/data/external_table_p0/paimon/paimon_timestamp_types.out
b/regression-test/data/external_table_p0/paimon/paimon_timestamp_types.out
index 32eaea02ed1..f4734e460f6 100644
--- a/regression-test/data/external_table_p0/paimon/paimon_timestamp_types.out
+++ b/regression-test/data/external_table_p0/paimon/paimon_timestamp_types.out
@@ -97,6 +97,12 @@
-- !ts9_ne_parquet --
1
+-- !c1_v1 --
+1 2024-01-02T10:04:05.100 2024-01-02T10:04:05.120 2024-01-02T10:04:05.123
2024-01-02T10:04:05.123400 2024-01-02T10:04:05.123450
2024-01-02T10:04:05.123456 2024-01-02T10:04:05.123456
2024-01-02T10:04:05.123456 2024-01-02T10:04:05.123456 2024-01-02
10:04:05.1+08:00 2024-01-02 10:04:05.12+08:00 2024-01-02
10:04:05.123+08:00 2024-01-02 10:04:05.1234+08:00 2024-01-02
10:04:05.12345+08:00 2024-01-02 10:04:05.123456+08:00 2024-01-02
10:04:05.123456+08:00 2024-01-02 10:04:05.123456+08:00 2024-01
[...]
+
+-- !c2_v1 --
+1 2024-01-02T10:04:05.100 2024-01-02T10:04:05.120 2024-01-02T10:04:05.123
2024-01-02T10:04:05.123400 2024-01-02T10:04:05.123450
2024-01-02T10:04:05.123456 2024-01-02T10:04:05.123456
2024-01-02T10:04:05.123456 2024-01-02T10:04:05.123456 2024-01-02
10:04:05.1+08:00 2024-01-02 10:04:05.12+08:00 2024-01-02
10:04:05.123+08:00 2024-01-02 10:04:05.1234+08:00 2024-01-02
10:04:05.12345+08:00 2024-01-02 10:04:05.123456+08:00 2024-01-02
10:04:05.123456+08:00 2024-01-02 10:04:05.123456+08:00 2024-01
[...]
+
-- !ltz_ntz_simple2 --
1 {"2024-01-01 10:12:34.123456":"2024-01-02 10:12:34.123456", "2024-01-03
10:12:34.123456":"2024-01-04 10:12:34.123456"} {"2024-01-03
10:12:34.123456+08:00":"2024-01-04 10:12:34.123456+08:00", "2024-01-01
10:12:34.123456+08:00":"2024-01-02 10:12:34.123456+08:00"} ["2024-01-01
10:12:34.123456", "2024-01-02 10:12:34.123456", "2024-01-03 10:12:34.123456"]
["2024-01-01 10:12:34.123456+08:00", "2024-01-02 10:12:34.123456+08:00",
"2024-01-03 10:12:34.123456+08:00"] {"crow1":"2024-01-01 10:12:3 [...]
diff --git
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_types.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_types.out
index 9125b18a52b..e4a254f9bf7 100644
---
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_types.out
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_types.out
@@ -26,12 +26,26 @@ event_time timestamptz(6) Yes true \N
WITH_TIMEZONE
-- !types_timezone_utc --
1 2024-01-15 02:30:00.123456+00:00
2 2024-01-15 02:30:00.654321+00:00
+3 1969-12-31 16:00:00.000000+00:00
-- !types_timezone_shanghai --
1 2024-01-15 10:30:00.123456+08:00
2 2024-01-15 10:30:00.654321+08:00
+3 1970-01-01 00:00:00.000000+08:00
-- !types_ntz --
1 2024-03-10T02:30:00.123456
2 2024-01-15T10:30:00.654321
+-- !types_fold_micros --
+1 1699173000123456
+2 1699176600123456
+
+-- !types_fold_jni_false --
+1 2023-11-05 08:30:00.123456+00:00
+2 2023-11-05 09:30:00.123456+00:00
+
+-- !types_fold_jni_true --
+1 2023-11-05 08:30:00.123456+00:00
+2 2023-11-05 09:30:00.123456+00:00
+
diff --git
a/regression-test/suites/external_table_p0/paimon/paimon_timestamp_types.groovy
b/regression-test/suites/external_table_p0/paimon/paimon_timestamp_types.groovy
index 118e79b0598..fe3f3155141 100644
---
a/regression-test/suites/external_table_p0/paimon/paimon_timestamp_types.groovy
+++
b/regression-test/suites/external_table_p0/paimon/paimon_timestamp_types.groovy
@@ -164,6 +164,11 @@ suite("paimon_timestamp_types", "p0,external") {
test_scale("2024-01-02 10:04:05.123457")
// Keep testing legacy LTZ fallback with V2 disabled independently of
the NTZ predicate tests.
sql """set enable_file_scanner_v2=false"""
+ // V1 must still decode every timestamp scale. Legacy ORC LTZ files
need Paimon's
+ // compatibility conversion even when JNI is not forced, while Parquet
stays on the
+ // native V1 reader. The ts9 predicates above stay on V2 only.
+ order_qt_c1_v1 """select * from ts_scale_orc"""
+ order_qt_c2_v1 """select * from ts_scale_parquet"""
// test_ltz_ntz("test_timestamp_ntz_ltz_orc")
// test_ltz_ntz("test_timestamp_ntz_ltz_parquet")
test_ltz_ntz_simple("test_timestamp_ntz_ltz_simple_orc", "2024-01-02
10:12:34.123456+08:00")
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_types.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_types.groovy
index a32f8810dd4..b9394e8bc79 100644
---
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_types.groovy
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_types.groovy
@@ -83,6 +83,11 @@ suite("test_paimon_write_types", "p0,external,paimon") {
CREATE TABLE paimon.${dbName}.t_types_ntz (
id INT, event_time TIMESTAMP_NTZ
) USING paimon;
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_types_fold;
+ CREATE TABLE paimon.${dbName}.t_types_fold (
+ id INT, event_time TIMESTAMP
+ ) USING paimon;
"""
sql """drop catalog if exists ${catalogName}"""
@@ -182,13 +187,19 @@ suite("test_paimon_write_types", "p0,external,paimon") {
sql """SET time_zone = 'Asia/Shanghai'"""
sql """INSERT INTO t_types_timezone VALUES
(1, TIMESTAMP '2024-01-15 10:30:00.123456')"""
+ // Shanghai's epoch midnight is a pre-epoch UTC instant.
+ sql """INSERT INTO t_types_timezone VALUES
+ (3, TIMESTAMP '1970-01-01 00:00:00')"""
sql """SET time_zone = 'UTC'"""
sql """INSERT INTO t_types_timezone VALUES
(2, TIMESTAMP '2024-01-15 02:30:00.654321')"""
order_qt_types_timezone_utc """SELECT id, event_time FROM
t_types_timezone ORDER BY id"""
+ // Spark reads the instants independently of the Doris reader.
+ assertTableEquals("t_types_timezone", "id, unix_micros(event_time)",
"ORDER BY id",
+ "id, " + epochMicros("event_time"))
// Reading the same snapshot in Asia/Shanghai must apply the session
offset
- // to both rows without losing their microsecond fractions.
+ // to every row without losing their microsecond fractions.
sql """SET time_zone = 'Asia/Shanghai'"""
order_qt_types_timezone_shanghai """SELECT id, event_time FROM
t_types_timezone ORDER BY id"""
@@ -204,6 +215,27 @@ suite("test_paimon_write_types", "p0,external,paimon") {
sql """SET time_zone = 'UTC'"""
order_qt_types_ntz """SELECT id, event_time FROM t_types_ntz ORDER BY
id"""
assertTableEquals("t_types_ntz", "*", "ORDER BY id")
+
+ // Both UTC instants render as 01:30 in Los Angeles. INSERT SELECT
must reach the
+ // LTZ writer without a lossy intermediate civil-time cast.
+ sql """SET time_zone = 'America/Los_Angeles'"""
+ sql """INSERT INTO t_types_fold
+ SELECT 1, CAST('2023-11-05 08:30:00.123456+00:00' AS
TIMESTAMPTZ(6))
+ UNION ALL
+ SELECT 2, CAST('2023-11-05 09:30:00.123456+00:00' AS
TIMESTAMPTZ(6))"""
+ sql """SET time_zone = 'UTC'"""
+ order_qt_types_fold_micros """SELECT id, ${epochMicros("event_time")}
FROM t_types_fold ORDER BY id"""
+ assertTableEquals("t_types_fold", "id, unix_micros(event_time)",
"ORDER BY id",
+ "id, " + epochMicros("event_time"))
+ def originalJniScanner = sql """SELECT @@force_jni_scanner"""
+ try {
+ for (boolean jni : [false, true]) {
+ sql """SET force_jni_scanner = ${jni}"""
+ "order_qt_types_fold_jni_${jni}" """SELECT id, event_time FROM
t_types_fold ORDER BY id"""
+ }
+ } finally {
+ sql """SET force_jni_scanner = ${originalJniScanner[0][0]}"""
+ }
} finally {
sql """SET time_zone = '${originalTimeZone[0][0]}'"""
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
