This is an automated email from the ASF dual-hosted git repository.
hello-stephen 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 13d18f3c88f [improvement](regression) Use Spark thrift JDBC for
external SQL helpers (#64886)
13d18f3c88f is described below
commit 13d18f3c88f5d083433366e6bd8382f4d4cf1d85
Author: zgxme <[email protected]>
AuthorDate: Fri Jul 3 10:47:03 2026 +0800
[improvement](regression) Use Spark thrift JDBC for external SQL helpers
(#64886)
### What problem does this PR solve?
This PR improves the execution efficiency of external SQL helper related
regression cases by using Spark thrift JDBC access.
Local validation shows the following cache-related cases are
significantly faster after this change:
- `test_iceberg_table_cache`: 3m20s -> 30s
- `test_paimon_table_meta_cache`: 14m59s -> 40s
---
.../docker-compose/iceberg/entrypoint.sh.tpl | 37 +-
.../docker-compose/iceberg/iceberg.env | 1 +
.../docker-compose/iceberg/iceberg.yaml.tpl | 2 +
.../create_preinstalled_scripts/paimon/run06.sql | 2 +-
.../docker-compose/iceberg/spark-defaults.conf | 3 +-
.../org/apache/doris/regression/suite/Suite.groovy | 135 ++--
.../doris/regression/suite/SuiteContext.groovy | 35 +
.../apache/doris/regression/util/JdbcUtils.groovy | 9 +-
.../doris/regression/util/ResultUtils.groovy | 756 +++++++++++++++++++++
...est_iceberg_spark_doris_consistency_demo.groovy | 301 ++++++++
...test_paimon_spark_doris_consistency_demo.groovy | 317 +++++++++
11 files changed, 1504 insertions(+), 94 deletions(-)
diff --git a/docker/thirdparties/docker-compose/iceberg/entrypoint.sh.tpl
b/docker/thirdparties/docker-compose/iceberg/entrypoint.sh.tpl
index 4232b4f3cc1..7c5dba39109 100644
--- a/docker/thirdparties/docker-compose/iceberg/entrypoint.sh.tpl
+++ b/docker/thirdparties/docker-compose/iceberg/entrypoint.sh.tpl
@@ -19,13 +19,14 @@
export SPARK_MASTER_HOST=doris--spark-iceberg
# wait iceberg-rest start
-while [[ ! $(curl -s --fail http://rest:8181/v1/config) ]]; do
+while ! curl -s --fail http://rest:8181/v1/config >/dev/null; do
sleep 1
done
set -ex
mkdir -p /opt/spark/events
+SPARK_THRIFT_EXTENSIONS="org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions"
for f in /opt/spark/sbin/*; do
ln -s $f /usr/local/bin/$(basename $f)
@@ -39,7 +40,6 @@ done
start-master.sh -p 7077
start-worker.sh spark://doris--spark-iceberg:7077
start-history-server.sh
-start-thriftserver.sh --driver-java-options "-Dderby.system.home=/tmp/derby"
# The creation of a Spark SQL client is time-consuming,
# and reopening a new client for each SQL file execution leads to significant
overhead.
@@ -68,6 +68,39 @@ END_TIME3=$(date +%s)
EXECUTION_TIME3=$((END_TIME3 - START_TIME3))
echo "Script iceberg load total: {} executed in $EXECUTION_TIME3 seconds"
+spark-sql \
+ --master spark://doris--spark-iceberg:7077 \
+ --conf
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
\
+ -e "CREATE DATABASE IF NOT EXISTS demo.default"
+
+start-thriftserver.sh \
+ --master spark://doris--spark-iceberg:7077 \
+ --conf "spark.sql.extensions=${SPARK_THRIFT_EXTENSIONS}" \
+ --conf spark.dynamicAllocation.enabled=false \
+ --conf spark.cores.max=8 \
+ --conf spark.executor.cores=4 \
+ --conf spark.executor.memory=1g \
+ --conf spark.driver.memory=1g \
+ --conf spark.sql.shuffle.partitions=16 \
+ --conf spark.default.parallelism=16 \
+ --driver-java-options "-Dderby.system.home=/tmp/derby"
+
+SPARK_THRIFT_READY_ATTEMPTS=0
+while ! beeline \
+ -u "jdbc:hive2://localhost:10000/default" \
+ -n hadoop \
+ -p hadoop \
+ -e "SELECT 1" >/tmp/spark-thriftserver-ready.log 2>&1; do
+ SPARK_THRIFT_READY_ATTEMPTS=$((SPARK_THRIFT_READY_ATTEMPTS + 1))
+ if [ "${SPARK_THRIFT_READY_ATTEMPTS}" -ge 120 ]; then
+ echo "ERROR: Spark thriftserver did not become ready after
${SPARK_THRIFT_READY_ATTEMPTS} attempts" >&2
+ cat /tmp/spark-thriftserver-ready.log >&2 || true
+ tail -n 200 /opt/spark/logs/*HiveThriftServer2*.out >&2 || true
+ exit 1
+ fi
+ sleep 1
+done
+
touch /mnt/SUCCESS;
tail -f /dev/null
diff --git a/docker/thirdparties/docker-compose/iceberg/iceberg.env
b/docker/thirdparties/docker-compose/iceberg/iceberg.env
index 6bebd49f437..0950783075c 100644
--- a/docker/thirdparties/docker-compose/iceberg/iceberg.env
+++ b/docker/thirdparties/docker-compose/iceberg/iceberg.env
@@ -19,6 +19,7 @@
NOTEBOOK_SERVER_PORT=8888
SPARK_DRIVER_UI_PORT=8080
SPARK_HISTORY_UI_PORT=10000
+SPARK_THRIFT_PORT=11000
REST_CATALOG_PORT=18181
MINIO_UI_PORT=9000
MINIO_API_PORT=19001
diff --git a/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl
b/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl
index 0c4e8e1cc65..9b1704a7891 100644
--- a/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl
+++ b/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl
@@ -41,6 +41,8 @@ services:
- AWS_ACCESS_KEY_ID=admin
- AWS_SECRET_ACCESS_KEY=password
- AWS_REGION=us-east-1
+ ports:
+ - ${SPARK_THRIFT_PORT}:10000
entrypoint: /bin/sh /mnt/scripts/entrypoint.sh
user: root
networks:
diff --git
a/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/paimon/run06.sql
b/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/paimon/run06.sql
index eb60255a08e..026bd8aab72 100644
---
a/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/paimon/run06.sql
+++
b/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/paimon/run06.sql
@@ -228,4 +228,4 @@ VALUES (1, NULL, 100.0),
(2, 'NULL', 200.0),
(3, '\\N', 300.0),
(4, 'null', 400.0),
- (5, 'A', 500.0);
\ No newline at end of file
+ (5, 'A', 500.0);
diff --git a/docker/thirdparties/docker-compose/iceberg/spark-defaults.conf
b/docker/thirdparties/docker-compose/iceberg/spark-defaults.conf
index 8336a2afcf8..f05bf40726f 100644
--- a/docker/thirdparties/docker-compose/iceberg/spark-defaults.conf
+++ b/docker/thirdparties/docker-compose/iceberg/spark-defaults.conf
@@ -20,6 +20,7 @@
# Example:
spark.sql.session.timeZone Asia/Shanghai
+
spark.sql.catalog.demo
org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.demo.type rest
spark.sql.catalog.demo.uri http://rest:8181
@@ -42,4 +43,4 @@ spark.sql.catalog.paimon.warehouse
s3://warehouse/wh
spark.sql.catalog.paimon.s3.endpoint http://minio:9000
spark.sql.catalog.paimon.s3.access-key admin
spark.sql.catalog.paimon.s3.secret-key password
-spark.sql.catalog.paimon.s3.region us-east-1
\ No newline at end of file
+spark.sql.catalog.paimon.s3.region us-east-1
diff --git
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
index 0dc8f9d1737..832669de082 100644
---
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
+++
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
@@ -53,6 +53,7 @@ import org.apache.doris.regression.util.DataUtils
import org.apache.doris.regression.util.JdbcUtils
import org.apache.doris.regression.util.Hdfs
import org.apache.doris.regression.util.Http
+import org.apache.doris.regression.util.ResultUtils
import org.apache.doris.regression.util.SuiteUtils
import org.apache.doris.regression.util.DebugPoint
import org.apache.doris.regression.RunMode
@@ -111,7 +112,6 @@ class Suite implements GroovyInterceptable {
private AmazonS3 s3Client = null
private FileSystem fs = null
- private String sparkIcebergContainerNameCache = null
Suite(String name, String group, SuiteContext context, SuiteCluster
cluster) {
this.name = name
@@ -1618,80 +1618,41 @@ class Suite implements GroovyInterceptable {
return result
}
- /**
- * Get the spark-iceberg container name by querying docker.
- * Uses 'docker ps --filter name=spark-iceberg' to find the container.
- */
- private String getSparkIcebergContainerName() {
- if (!Strings.isNullOrEmpty(sparkIcebergContainerNameCache)) {
- return sparkIcebergContainerNameCache
- }
-
- try {
- // Use docker ps with filter to find containers with
'spark-iceberg' in the name
- String command = "docker ps --filter name=spark-iceberg --format
{{.Names}}"
- def process = command.execute()
- process.waitFor()
- String output = process.in.text.trim()
+ private List<List<Object>> spark_sql(String sqlStr, boolean isOrder =
false) {
+ String cleanedSqlStr = sqlStr.replaceAll("\\s*;\\s*\$", "")
+ logger.info("Execute Spark JDBC SQL: ${cleanedSqlStr}".toString())
+ logger.info("Spark JDBC URL:
${context.getSparkIcebergJdbcUrl()}".toString())
+ return sql_impl(context.getSparkIcebergConnection(), cleanedSqlStr,
isOrder)
+ }
- if (output) {
- // Get the first matching container
- String containerName = output.split('\n')[0].trim()
- if (containerName) {
- sparkIcebergContainerNameCache = containerName
- logger.info("Found spark-iceberg container:
${containerName}".toString())
- return containerName
- }
- }
+ private List spark_sql_multi(Object sqlStatements, boolean isOrder =
false) {
+ def statements = sqlStatements.toString().split(';').collect {
it.trim() }.findAll { it }
- logger.warn("No spark-iceberg container found via docker ps")
- return null
- } catch (Exception e) {
- logger.warn("Failed to get spark-iceberg container via docker ps:
${e.message}".toString())
- return null
+ if (statements.isEmpty()) {
+ return []
}
+
+ logger.info("Execute Spark JDBC SQL statements via
${context.getSparkIcebergJdbcUrl()}: ${statements}".toString())
+ Connection sparkConn = context.getSparkIcebergConnection()
+ return statements.collect { statement -> sql_impl(sparkConn,
statement, isOrder) }
}
/**
- * Execute Spark SQL on the spark-iceberg container via docker exec.
+ * Execute Spark SQL on the Spark ThriftServer via Hive JDBC.
*
* Usage in test suite:
* spark_iceberg "CREATE TABLE demo.test_db.t1 (id INT) USING iceberg"
* spark_iceberg "INSERT INTO demo.test_db.t1 VALUES (1)"
* def result = spark_iceberg "SELECT * FROM demo.test_db.t1"
- *
- * The container name is found by querying 'docker ps --filter
name=spark-iceberg'
*/
- String spark_iceberg(String sqlStr, int timeoutSeconds = 120) {
- String containerName = getSparkIcebergContainerName()
- if (containerName == null) {
- throw new RuntimeException("spark-iceberg container not found.
Please ensure the container is running.")
- }
- String masterUrl = "spark://${containerName}:7077"
-
- // Escape double quotes in SQL string for shell command
- String escapedSql = sqlStr.replaceAll('"', '\\\\"')
-
- // Build docker exec command
- String command = """docker exec ${containerName} spark-sql --master
${masterUrl} --conf
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
-e "${escapedSql}" """
-
- logger.info("Executing Spark Iceberg SQL: ${sqlStr}".toString())
- logger.info("Container: ${containerName}".toString())
-
- try {
- String result = cmd(command, timeoutSeconds)
- logger.info("Spark Iceberg SQL result: ${result}".toString())
- return result
- } catch (Exception e) {
- logger.error("Spark Iceberg SQL failed: ${e.message}".toString())
- throw e
- }
+ List<List<Object>> spark_iceberg(String sqlStr, boolean isOrder = false) {
+ return spark_sql(sqlStr, isOrder)
}
/**
- * Execute multiple Spark SQL statements on the spark-iceberg container.
+ * Execute multiple Spark SQL statements on the Spark ThriftServer via
Hive JDBC.
* Statements are separated by semicolons.
- * All statements are executed in one spark-sql process to reduce startup
overhead.
+ * All statements are executed on one JDBC connection to reduce startup
overhead.
*
* Usage:
* spark_iceberg_multi '''
@@ -1700,46 +1661,40 @@ class Suite implements GroovyInterceptable {
* INSERT INTO demo.test_db.t1 VALUES (1);
* '''
*/
- List<String> spark_iceberg_multi(String sqlStatements, int timeoutSeconds
= 300) {
- def statements = sqlStatements.split(';').collect { it.trim()
}.findAll { it }
-
- if (statements.isEmpty()) {
- return []
- }
-
- String combinedSql = statements.collect { "${it};" }.join(" ")
- return [spark_iceberg(combinedSql, timeoutSeconds)]
+ List spark_iceberg_multi(Object sqlStatements, boolean isOrder = false) {
+ return spark_sql_multi(sqlStatements, isOrder)
}
/**
- * Execute Spark SQL on the spark-iceberg container with Paimon extensions
enabled.
+ * Execute Spark SQL with the Paimon catalog on the Spark ThriftServer via
Hive JDBC.
*
* Usage in test suite:
* spark_paimon "CREATE TABLE paimon.test_db.t1 (id INT) USING paimon"
* spark_paimon "INSERT INTO paimon.test_db.t1 VALUES (1)"
* def result = spark_paimon "SELECT * FROM paimon.test_db.t1"
*/
- String spark_paimon(String sqlStr, int timeoutSeconds = 120) {
- String containerName = getSparkIcebergContainerName()
- if (containerName == null) {
- throw new RuntimeException("spark-iceberg container not found.
Please ensure the container is running.")
- }
- String masterUrl = "spark://${containerName}:7077"
-
- String escapedSql = sqlStr.replaceAll('"', '\\\\"')
- String command = """docker exec ${containerName} spark-sql --master
${masterUrl} --conf
spark.sql.extensions=org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
-e "${escapedSql}" """
+ List<List<Object>> spark_paimon(String sqlStr, boolean isOrder = false) {
+ return spark_sql(sqlStr, isOrder)
+ }
- logger.info("Executing Spark Paimon SQL: ${sqlStr}".toString())
- logger.info("Container: ${containerName}".toString())
+ /**
+ * Execute multiple Spark SQL statements with the Paimon catalog on the
Spark ThriftServer via Hive JDBC.
+ * Statements are separated by semicolons.
+ * All statements are executed on one JDBC connection to reduce startup
overhead.
+ *
+ * Usage:
+ * spark_paimon_multi '''
+ * CREATE DATABASE IF NOT EXISTS paimon.test_db;
+ * CREATE TABLE paimon.test_db.t1 (id INT) USING paimon;
+ * INSERT INTO paimon.test_db.t1 VALUES (1);
+ * '''
+ */
+ List spark_paimon_multi(Object sqlStatements, boolean isOrder = false) {
+ return spark_sql_multi(sqlStatements, isOrder)
+ }
- try {
- String result = cmd(command, timeoutSeconds)
- logger.info("Spark Paimon SQL result: ${result}".toString())
- return result
- } catch (Exception e) {
- logger.error("Spark Paimon SQL failed: ${e.message}".toString())
- throw e
- }
+ void assertSparkDorisResultEquals(List<List<Object>> sparkRows,
List<List<Object>> dorisRows) {
+ ResultUtils.assertSparkDorisResultEquals(sparkRows, dorisRows)
}
List<List<Object>> db2_docker(String sqlStr, boolean isOrder = false) {
@@ -1970,6 +1925,10 @@ class Suite implements GroovyInterceptable {
return quickTest(name.substring("order_qt_".length()), (args as
Object[])[0] as String, true)
} else if (name.startsWith("qe_")) {
return quickExecute(name.substring("qe_".length()), (args as
Object[])[0] as PreparedStatement)
+ } else if (name == "assertSparkDorisResultEquals") {
+ ResultUtils.assertSparkDorisResultEquals((args as Object[])[0] as
List<List<Object>>,
+ (args as Object[])[1] as List<List<Object>>)
+ return null
} else if (name.startsWith("assert") && name.length() >
"assert".length()) {
// delegate to junit Assertions dynamically
return Assertions."$name"(*args) // *args: spread-dot
diff --git
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteContext.groovy
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteContext.groovy
index 0d599aed817..fcd59cb3a73 100644
---
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteContext.groovy
+++
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteContext.groovy
@@ -53,6 +53,7 @@ class SuiteContext implements Closeable {
public final ThreadLocal<Connection> threadHive2DockerConn = new
ThreadLocal<>()
public final ThreadLocal<Connection> threadHive3DockerConn = new
ThreadLocal<>()
public final ThreadLocal<Connection> threadHiveRemoteConn = new
ThreadLocal<>()
+ public final ThreadLocal<Connection> threadSparkIcebergConn = new
ThreadLocal<>()
public final ThreadLocal<Connection> threadDB2DockerConn = new
ThreadLocal<>()
private final ThreadLocal<Syncer> syncer = new ThreadLocal<>()
public final Config config
@@ -239,6 +240,15 @@ class SuiteContext implements Closeable {
return threadConn
}
+ Connection getSparkIcebergConnection() {
+ def threadConn = threadSparkIcebergConn.get()
+ if (threadConn == null) {
+ threadConn = getConnectionBySparkIcebergConfig()
+ threadSparkIcebergConn.set(threadConn)
+ }
+ return threadConn
+ }
+
Connection getDB2DockerConnection() {
def threadConn = threadDB2DockerConn.get()
if (threadConn == null) {
@@ -314,6 +324,21 @@ class SuiteContext implements Closeable {
return DriverManager.getConnection(hiveJdbcUrl, hiveJdbcUser,
hiveJdbcPassword)
}
+ Connection getConnectionBySparkIcebergConfig() {
+ Class.forName("org.apache.hive.jdbc.HiveDriver");
+ String sparkJdbcUser = "hadoop"
+ String sparkJdbcPassword = "hadoop"
+ String sparkJdbcUrl = getSparkIcebergJdbcUrl()
+ log.info("Create Spark Iceberg JDBC connection to
${sparkJdbcUrl}".toString())
+ return DriverManager.getConnection(sparkJdbcUrl, sparkJdbcUser,
sparkJdbcPassword)
+ }
+
+ String getSparkIcebergJdbcUrl() {
+ String sparkHost = config.otherConfigs.get("externalEnvIp")
+ String sparkPort =
config.otherConfigs.get("iceberg_spark_thrift_port") ?: "11000"
+ return "jdbc:hive2://${sparkHost}:${sparkPort}/default"
+ }
+
Connection getConnectionByDB2DockerConfig() {
Class.forName("com.ibm.db2.jcc.DB2Driver");
String db2Host = config.otherConfigs.get("externalEnvIp")
@@ -616,6 +641,16 @@ class SuiteContext implements Closeable {
log.warn("Close connection failed", t)
}
}
+
+ Connection spark_iceberg_conn = threadSparkIcebergConn.get()
+ if (spark_iceberg_conn != null) {
+ threadSparkIcebergConn.remove()
+ try {
+ spark_iceberg_conn.close()
+ } catch (Throwable t) {
+ log.warn("Close connection failed", t)
+ }
+ }
}
diff --git
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/JdbcUtils.groovy
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/JdbcUtils.groovy
index bac8a2087d5..c79303c6f1e 100644
---
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/JdbcUtils.groovy
+++
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/JdbcUtils.groovy
@@ -24,6 +24,7 @@ import java.sql.Connection
import java.sql.PreparedStatement
import java.sql.ResultSet
import java.sql.ResultSetMetaData
+import java.sql.SQLFeatureNotSupportedException
import java.sql.Types
import org.slf4j.Logger
import org.slf4j.LoggerFactory
@@ -126,8 +127,12 @@ class JdbcUtils {
for (int i = 1; i <= columnCount; ++i) {
int jdbcType = resultSet.metaData.getColumnType(i)
if (isBinaryJdbcType(jdbcType)) {
- byte[] bytes = resultSet.getBytes(i)
- row.add(bytes == null ? null : bytesToHex(bytes))
+ try {
+ byte[] bytes = resultSet.getBytes(i)
+ row.add(bytes == null ? null : bytesToHex(bytes))
+ } catch (SQLFeatureNotSupportedException ignored) {
+ row.add(resultSet.getObject(i))
+ }
} else {
row.add(resultSet.getObject(i))
}
diff --git
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/ResultUtils.groovy
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/ResultUtils.groovy
new file mode 100644
index 00000000000..897377b8e9d
--- /dev/null
+++
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/ResultUtils.groovy
@@ -0,0 +1,756 @@
+// 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.regression.util
+
+import groovy.transform.CompileStatic
+import org.junit.Assert
+
+import java.math.BigDecimal
+import java.math.BigInteger
+import java.nio.ByteBuffer
+import java.sql.Blob
+import java.sql.Clob
+import java.sql.Struct
+import java.sql.Time
+import java.sql.Timestamp
+import java.time.temporal.TemporalAccessor
+import java.util.regex.Pattern
+
+@CompileStatic
+class ResultUtils {
+ private static final Object PARSE_FAILED = new Object()
+ private static final double NUMERIC_RELATIVE_TOLERANCE = 1.0E-8D
+ private static final Pattern NUMBER_PATTERN = Pattern.compile(
+ "[-+]?(?:\\d+(?:\\.\\d*)?|\\.\\d+)(?:[eE][-+]?\\d+)?")
+ private static final Pattern PREFIXED_HEX_PATTERN =
Pattern.compile("(?i)0x[0-9a-f]*")
+ private static final Pattern BARE_HEX_PATTERN =
Pattern.compile("(?i)[0-9a-f]*")
+ private static final Pattern DATE_PATTERN =
Pattern.compile("\\d{4}-\\d{2}-\\d{2}")
+ private static final Pattern TIME_PATTERN = Pattern.compile(
+ "(\\d{2}:\\d{2})(?::(\\d{2})(?:\\.(\\d{1,9}))?)?")
+ private static final Pattern DATETIME_PATTERN = Pattern.compile(
+ "(\\d{4}-\\d{2}-\\d{2})[T
](\\d{2}:\\d{2})(?::(\\d{2})(?:\\.(\\d{1,9}))?)?")
+ private static final Pattern UNICODE_ESCAPE_PATTERN =
Pattern.compile("[0-9a-fA-F]{4}")
+
+ static List<List<Object>> normalizeRows(List<List<Object>> rows) {
+ List<List<Object>> normalizedRows = new ArrayList<>()
+ for (List<Object> row : rows) {
+ List<Object> normalizedRow = new ArrayList<>()
+ for (Object value : row) {
+ normalizedRow.add(normalizeValue(value))
+ }
+ normalizedRows.add(normalizedRow)
+ }
+ return normalizedRows
+ }
+
+ static void assertSparkDorisResultEquals(List<List<Object>> sparkRows,
List<List<Object>> dorisRows) {
+ List<List<Object>> normalizedSparkRows = normalizeRows(sparkRows)
+ List<List<Object>> normalizedDorisRows = normalizeRows(dorisRows)
+ Assert.assertEquals("Spark and Doris result row count mismatch",
+ normalizedSparkRows.size(), normalizedDorisRows.size())
+
+ for (int rowIndex = 0; rowIndex < normalizedSparkRows.size();
rowIndex++) {
+ List<Object> sparkRow = normalizedSparkRows.get(rowIndex)
+ List<Object> dorisRow = normalizedDorisRows.get(rowIndex)
+ Assert.assertEquals("Spark and Doris result column count mismatch
at row " + (rowIndex + 1),
+ sparkRow.size(), dorisRow.size())
+ for (int columnIndex = 0; columnIndex < sparkRow.size();
columnIndex++) {
+ Object sparkValue = sparkRow.get(columnIndex)
+ Object dorisValue = dorisRow.get(columnIndex)
+ if (!valueEquals(sparkValue, dorisValue)) {
+ Assert.fail("Spark/Doris result mismatch at row " +
(rowIndex + 1)
+ + ", column " + (columnIndex + 1)
+ + "\nSpark raw : " +
valueToString(sparkRows.get(rowIndex).get(columnIndex))
+ + "\nDoris raw : " +
valueToString(dorisRows.get(rowIndex).get(columnIndex))
+ + "\nSpark normalized: " +
valueToString(sparkValue)
+ + "\nDoris normalized: " +
valueToString(dorisValue))
+ }
+ }
+ }
+ }
+
+ static Object normalizeValue(Object value) {
+ if (value == null) {
+ return null
+ }
+ if (value instanceof byte[]) {
+ return bytesToHex((byte[]) value)
+ }
+ if (value instanceof ByteBuffer) {
+ return bytesToHex(byteBufferToBytes((ByteBuffer) value))
+ }
+ if (value instanceof BigDecimal) {
+ return normalizeDecimal((BigDecimal) value)
+ }
+ if (value instanceof BigInteger
+ || value instanceof Byte
+ || value instanceof Short
+ || value instanceof Integer
+ || value instanceof Long) {
+ return normalizeDecimal(new BigDecimal(value.toString()))
+ }
+ if (value instanceof Float || value instanceof Double) {
+ Double doubleValue = ((Number) value).doubleValue()
+ if (doubleValue.isNaN() || doubleValue.isInfinite()) {
+ return doubleValue.toString().toLowerCase(Locale.ROOT)
+ }
+ return new ApproximateNumber(normalizeDecimal(new
BigDecimal(value.toString())))
+ }
+ if (value instanceof Boolean) {
+ return value
+ }
+ if (value instanceof java.sql.Array) {
+ try {
+ return normalizeValue(((java.sql.Array) value).getArray())
+ } catch (Exception ignored) {
+ return value.toString()
+ }
+ }
+ if (value instanceof Struct) {
+ try {
+ return normalizeValue(Arrays.asList(((Struct)
value).getAttributes()))
+ } catch (Exception ignored) {
+ return value.toString()
+ }
+ }
+ if (value instanceof Blob) {
+ Blob blob = (Blob) value
+ try {
+ return bytesToHex(blob.getBytes(1L, (int) blob.length()))
+ } catch (Exception ignored) {
+ return value.toString()
+ }
+ }
+ if (value instanceof Clob) {
+ Clob clob = (Clob) value
+ try {
+ return normalizeValue(clob.getSubString(1L, (int)
clob.length()))
+ } catch (Exception ignored) {
+ return value.toString()
+ }
+ }
+ if (value instanceof java.sql.Date
+ || value instanceof Time
+ || value instanceof Timestamp
+ || value instanceof java.util.Date
+ || value instanceof Calendar
+ || value instanceof TemporalAccessor) {
+ return normalizeString(value.toString())
+ }
+ if (value instanceof Map) {
+ TreeMap<String, Object> normalizedMap = new TreeMap<>()
+ for (Map.Entry<?, ?> entry : ((Map<?, ?>) value).entrySet()) {
+ normalizedMap.put(canonicalKey(entry.getKey()),
normalizeValue(entry.getValue()))
+ }
+ return normalizedMap
+ }
+ if (value instanceof Collection) {
+ List<Object> normalizedList = new ArrayList<>()
+ for (Object item : (Collection<?>) value) {
+ normalizedList.add(normalizeValue(item))
+ }
+ return normalizedList
+ }
+ if (value.getClass().isArray()) {
+ List<Object> normalizedList = new ArrayList<>()
+ int length = java.lang.reflect.Array.getLength(value)
+ for (int i = 0; i < length; i++) {
+
normalizedList.add(normalizeValue(java.lang.reflect.Array.get(value, i)))
+ }
+ return normalizedList
+ }
+ if (value instanceof CharSequence) {
+ return normalizeString(value.toString())
+ }
+ return normalizeString(value.toString())
+ }
+
+ private static Object normalizeString(String raw) {
+ String text = raw.trim()
+ Object parsed = parseComplexValue(text)
+ if (parsed != PARSE_FAILED) {
+ return normalizeValue(parsed)
+ }
+
+ parsed = parseQuotedStringValue(text)
+ if (parsed != PARSE_FAILED) {
+ return normalizeValue(parsed)
+ }
+
+ if (PREFIXED_HEX_PATTERN.matcher(text).matches()) {
+ return normalizePrefixedHex(text)
+ }
+
+ String normalizedTemporal = normalizeTemporal(text)
+ if (normalizedTemporal != null) {
+ return normalizedTemporal
+ }
+
+ String lowerText = text.toLowerCase(Locale.ROOT)
+ if (lowerText == "nan"
+ || lowerText == "-nan"
+ || lowerText == "inf"
+ || lowerText == "+inf"
+ || lowerText == "-inf"
+ || lowerText == "infinity"
+ || lowerText == "+infinity"
+ || lowerText == "-infinity") {
+ return lowerText
+ }
+ return raw
+ }
+
+ private static Object parseQuotedStringValue(String text) {
+ if (text.length() < 2 || text.charAt(0) != ('"' as char)
+ || text.charAt(text.length() - 1) != ('"' as char)) {
+ return PARSE_FAILED
+ }
+ try {
+ return new ComplexValueParser(text).parse()
+ } catch (RuntimeException ignored) {
+ return PARSE_FAILED
+ }
+ }
+
+ private static Object parseComplexValue(String text) {
+ if (text.length() < 2) {
+ return PARSE_FAILED
+ }
+ char first = text.charAt(0)
+ char last = text.charAt(text.length() - 1)
+ if (!((first == '[' as char && last == ']' as char)
+ || (first == '{' as char && last == '}' as char))) {
+ return PARSE_FAILED
+ }
+ try {
+ return new ComplexValueParser(text).parse()
+ } catch (RuntimeException ignored) {
+ return PARSE_FAILED
+ }
+ }
+
+ private static String normalizeTemporal(String text) {
+ if (DATE_PATTERN.matcher(text).matches()) {
+ return text
+ }
+
+ def datetimeMatcher = DATETIME_PATTERN.matcher(text)
+ if (datetimeMatcher.matches()) {
+ return datetimeMatcher.group(1) + " " + normalizeTime(
+ datetimeMatcher.group(2), datetimeMatcher.group(3),
datetimeMatcher.group(4))
+ }
+
+ def timeMatcher = TIME_PATTERN.matcher(text)
+ if (timeMatcher.matches()) {
+ return normalizeTime(timeMatcher.group(1), timeMatcher.group(2),
timeMatcher.group(3))
+ }
+
+ return null
+ }
+
+ private static String normalizeTime(String hourMinute, String secondPart,
String fractionPart) {
+ String second = secondPart == null ? "00" : secondPart
+ String fraction = fractionPart
+ if (fraction != null) {
+ fraction = fraction.replaceAll("0+\$", "")
+ }
+ String suffix = fraction == null || fraction.isEmpty() ? "" : "." +
fraction
+ return hourMinute + ":" + second + suffix
+ }
+
+ private static String normalizePrefixedHex(String text) {
+ return "0x" + text.substring(2).toUpperCase(Locale.ROOT)
+ }
+
+ private static String bytesToHex(byte[] bytes) {
+ StringBuilder builder = new StringBuilder(2 + bytes.length * 2)
+ builder.append("0x")
+ for (byte b : bytes) {
+ builder.append(String.format("%02X", b & 0xFF))
+ }
+ return builder.toString()
+ }
+
+ private static byte[] byteBufferToBytes(ByteBuffer buffer) {
+ ByteBuffer copy = buffer.asReadOnlyBuffer()
+ byte[] bytes = new byte[copy.remaining()]
+ copy.get(bytes)
+ return bytes
+ }
+
+ private static BigDecimal normalizeDecimal(BigDecimal decimal) {
+ BigDecimal normalized = decimal.stripTrailingZeros()
+ if (normalized.compareTo(BigDecimal.ZERO) == 0) {
+ return BigDecimal.ZERO
+ }
+ return normalized
+ }
+
+ private static boolean valueEquals(Object left, Object right) {
+ if (left == null || right == null) {
+ return left == right
+ }
+ if (left.equals(right)) {
+ return true
+ }
+ if (left instanceof Map && right instanceof Map) {
+ return mapEquals((Map<?, ?>) left, (Map<?, ?>) right)
+ }
+ if (left instanceof List && right instanceof List) {
+ List<?> leftList = (List<?>) left
+ List<?> rightList = (List<?>) right
+ if (leftList.size() != rightList.size()) {
+ return false
+ }
+ for (int i = 0; i < leftList.size(); i++) {
+ if (!valueEquals(leftList.get(i), rightList.get(i))) {
+ return false
+ }
+ }
+ return true
+ }
+
+ if (hexEquals(left, right)) {
+ return true
+ }
+
+ if (left instanceof Boolean || right instanceof Boolean) {
+ Boolean leftBoolean = asBoolean(left)
+ Boolean rightBoolean = asBoolean(right)
+ if (leftBoolean != null && rightBoolean != null) {
+ return leftBoolean == rightBoolean
+ }
+ }
+
+ BigDecimal leftDecimal = asDecimal(left)
+ BigDecimal rightDecimal = asDecimal(right)
+ if (leftDecimal != null && rightDecimal != null) {
+ return numericEquals(left, right, leftDecimal, rightDecimal)
+ }
+ return left.equals(right)
+ }
+
+ private static boolean mapEquals(Map<?, ?> leftMap, Map<?, ?> rightMap) {
+ if (leftMap.size() != rightMap.size()) {
+ return false
+ }
+ if (leftMap.keySet().equals(rightMap.keySet())) {
+ for (Object key : leftMap.keySet()) {
+ if (!valueEquals(leftMap.get(key), rightMap.get(key))) {
+ return false
+ }
+ }
+ return true
+ }
+
+ Set<Object> matchedRightKeys = new HashSet<>()
+ for (Object leftKey : leftMap.keySet()) {
+ Object matchedRightKey = findMatchedMapKey(leftKey,
leftMap.get(leftKey), rightMap, matchedRightKeys)
+ if (matchedRightKey == null) {
+ return false
+ }
+ matchedRightKeys.add(matchedRightKey)
+ }
+ return true
+ }
+
+ private static Object findMatchedMapKey(Object leftKey, Object leftValue,
Map<?, ?> rightMap,
+ Set<Object> matchedRightKeys) {
+ for (Object rightKey : rightMap.keySet()) {
+ if (matchedRightKeys.contains(rightKey)
+ || !mapKeyEquals(leftKey, rightKey)
+ || !valueEquals(leftValue, rightMap.get(rightKey))) {
+ continue
+ }
+ return rightKey
+ }
+ return null
+ }
+
+ private static boolean mapKeyEquals(Object leftKey, Object rightKey) {
+ if (leftKey == null || rightKey == null) {
+ return leftKey == rightKey
+ }
+ if (leftKey.equals(rightKey)) {
+ return true
+ }
+ Boolean leftBoolean = booleanEquivalentMapKey(leftKey)
+ Boolean rightBoolean = booleanEquivalentMapKey(rightKey)
+ return leftBoolean != null && rightBoolean != null && leftBoolean ==
rightBoolean
+ }
+
+ private static Boolean booleanEquivalentMapKey(Object key) {
+ if (key instanceof CharSequence) {
+ String text = key.toString()
+ if (text == "bool:true") {
+ return true
+ }
+ if (text == "bool:false") {
+ return false
+ }
+ if (text.startsWith("num:")) {
+ return booleanFromDecimalText(text.substring("num:".length()))
+ }
+ return null
+ }
+ return asBoolean(key)
+ }
+
+ private static Boolean booleanFromDecimalText(String text) {
+ try {
+ BigDecimal decimal = normalizeDecimal(new BigDecimal(text))
+ if (decimal.compareTo(BigDecimal.ZERO) == 0) {
+ return false
+ }
+ if (decimal.compareTo(BigDecimal.ONE) == 0) {
+ return true
+ }
+ return null
+ } catch (NumberFormatException ignored) {
+ return null
+ }
+ }
+
+ private static boolean hexEquals(Object left, Object right) {
+ String leftHex = asHex(left, right)
+ String rightHex = asHex(right, left)
+ return leftHex != null && rightHex != null && leftHex == rightHex
+ }
+
+ private static String asHex(Object value, Object counterpart) {
+ if (!(value instanceof CharSequence) && !(value instanceof
BigDecimal)) {
+ return null
+ }
+ String text = value instanceof BigDecimal ? ((BigDecimal)
value).toPlainString() : value.toString().trim()
+ if (PREFIXED_HEX_PATTERN.matcher(text).matches()) {
+ return text.substring(2).toUpperCase(Locale.ROOT)
+ }
+ if (!(counterpart instanceof CharSequence)
+ ||
!PREFIXED_HEX_PATTERN.matcher(counterpart.toString().trim()).matches()
+ || text.length() % 2 != 0
+ || !BARE_HEX_PATTERN.matcher(text).matches()) {
+ return null
+ }
+ return text.toUpperCase(Locale.ROOT)
+ }
+
+ private static Boolean asBoolean(Object value) {
+ if (value instanceof Boolean) {
+ return (Boolean) value
+ }
+ BigDecimal decimal = asDecimal(value)
+ if (decimal != null) {
+ if (decimal.compareTo(BigDecimal.ZERO) == 0) {
+ return false
+ }
+ if (decimal.compareTo(BigDecimal.ONE) == 0) {
+ return true
+ }
+ return null
+ }
+ if (value instanceof CharSequence) {
+ String text = value.toString().trim().toLowerCase(Locale.ROOT)
+ if (text == "true") {
+ return true
+ }
+ if (text == "false") {
+ return false
+ }
+ }
+ return null
+ }
+
+ private static BigDecimal asDecimal(Object value) {
+ if (value instanceof BigDecimal) {
+ return (BigDecimal) value
+ }
+ if (value instanceof BigInteger
+ || value instanceof Byte
+ || value instanceof Short
+ || value instanceof Integer
+ || value instanceof Long) {
+ return normalizeDecimal(new BigDecimal(value.toString()))
+ }
+ if (value instanceof Float || value instanceof Double) {
+ Double doubleValue = ((Number) value).doubleValue()
+ if (doubleValue.isNaN() || doubleValue.isInfinite()) {
+ return null
+ }
+ return normalizeDecimal(new BigDecimal(value.toString()))
+ }
+ if (value instanceof ApproximateNumber) {
+ return ((ApproximateNumber) value).decimal
+ }
+ if (value instanceof CharSequence) {
+ String text = value.toString().trim()
+ if (NUMBER_PATTERN.matcher(text).matches()) {
+ return normalizeDecimal(new BigDecimal(text))
+ }
+ }
+ return null
+ }
+
+ private static boolean numericEquals(Object left, Object right, BigDecimal
leftDecimal, BigDecimal rightDecimal) {
+ if (left instanceof CharSequence && right instanceof CharSequence) {
+ return false
+ }
+ if (left instanceof ApproximateNumber || right instanceof
ApproximateNumber
+ || left instanceof Float || left instanceof Double
+ || right instanceof Float || right instanceof Double) {
+ return approximateDecimalEquals(leftDecimal, rightDecimal)
+ }
+ return leftDecimal.compareTo(rightDecimal) == 0
+ }
+
+ private static boolean approximateDecimalEquals(BigDecimal left,
BigDecimal right) {
+ if (left.compareTo(right) == 0) {
+ return true
+ }
+ double leftDouble = left.doubleValue()
+ double rightDouble = right.doubleValue()
+ if (Double.isInfinite(leftDouble) || Double.isInfinite(rightDouble)
+ || Double.isNaN(leftDouble) || Double.isNaN(rightDouble)) {
+ return false
+ }
+ double diff = Math.abs(leftDouble - rightDouble)
+ double base = Math.max(1.0D, Math.max(Math.abs(leftDouble),
Math.abs(rightDouble)))
+ return diff / base <= NUMERIC_RELATIVE_TOLERANCE
+ }
+
+ private static String canonicalKey(Object key) {
+ Object normalizedKey = normalizeValue(key)
+ if (normalizedKey instanceof Boolean) {
+ return ((Boolean) normalizedKey) ? "bool:true" : "bool:false"
+ }
+ if (!(normalizedKey instanceof CharSequence)) {
+ BigDecimal decimal = asDecimal(normalizedKey)
+ if (decimal != null) {
+ return "num:" + decimal.toPlainString()
+ }
+ }
+ if (normalizedKey instanceof Map || normalizedKey instanceof List) {
+ return "complex:" + valueToString(normalizedKey)
+ }
+ return "str:" + normalizedKey.toString()
+ }
+
+ private static String valueToString(Object value) {
+ if (value == null) {
+ return "null"
+ }
+ if (value instanceof BigDecimal) {
+ return ((BigDecimal) value).toPlainString()
+ }
+ if (value instanceof ApproximateNumber) {
+ return ((ApproximateNumber) value).decimal.toPlainString()
+ }
+ return value.toString()
+ }
+
+ private static final class ApproximateNumber {
+ private final BigDecimal decimal
+
+ private ApproximateNumber(BigDecimal decimal) {
+ this.decimal = decimal
+ }
+
+ @Override
+ String toString() {
+ return decimal.toPlainString()
+ }
+ }
+
+ private static final class ComplexValueParser {
+ private final String text
+ private int position = 0
+
+ ComplexValueParser(String text) {
+ this.text = text
+ }
+
+ Object parse() {
+ Object value = parseValue(false)
+ skipWhitespace()
+ if (position != text.length()) {
+ throw new IllegalArgumentException("Unexpected trailing
content")
+ }
+ return value
+ }
+
+ private Object parseValue(boolean stopAtColon) {
+ skipWhitespace()
+ if (position >= text.length()) {
+ throw new IllegalArgumentException("Unexpected end of input")
+ }
+ char ch = text.charAt(position)
+ if (ch == '[' as char) {
+ return parseArray()
+ }
+ if (ch == '{' as char) {
+ return parseObject()
+ }
+ if (ch == '"' as char) {
+ return parseQuotedString()
+ }
+ return parseToken(stopAtColon)
+ }
+
+ private List<Object> parseArray() {
+ position++
+ List<Object> values = new ArrayList<>()
+ skipWhitespace()
+ if (consume(']' as char)) {
+ return values
+ }
+ while (true) {
+ values.add(parseValue(false))
+ skipWhitespace()
+ if (consume(',' as char)) {
+ continue
+ }
+ if (consume(']' as char)) {
+ return values
+ }
+ throw new IllegalArgumentException("Expected ',' or ']'")
+ }
+ }
+
+ private Map<Object, Object> parseObject() {
+ position++
+ Map<Object, Object> values = new LinkedHashMap<>()
+ skipWhitespace()
+ if (consume('}' as char)) {
+ return values
+ }
+ while (true) {
+ Object key = parseValue(true)
+ skipWhitespace()
+ if (!consume(':' as char)) {
+ throw new IllegalArgumentException("Expected ':'")
+ }
+ Object value = parseValue(false)
+ values.put(key, value)
+ skipWhitespace()
+ if (consume(',' as char)) {
+ continue
+ }
+ if (consume('}' as char)) {
+ return values
+ }
+ throw new IllegalArgumentException("Expected ',' or '}'")
+ }
+ }
+
+ private String parseQuotedString() {
+ position++
+ StringBuilder builder = new StringBuilder()
+ while (position < text.length()) {
+ char ch = text.charAt(position++)
+ if (ch == '"' as char) {
+ return builder.toString()
+ }
+ if (ch == '\\' as char && position < text.length()) {
+ char escaped = text.charAt(position++)
+ switch (escaped) {
+ case '"' as char:
+ case '\\' as char:
+ case '/' as char:
+ builder.append(escaped)
+ break
+ case 'b' as char:
+ builder.append('\b' as char)
+ break
+ case 'f' as char:
+ builder.append('\f' as char)
+ break
+ case 'n' as char:
+ builder.append('\n' as char)
+ break
+ case 'r' as char:
+ builder.append('\r' as char)
+ break
+ case 't' as char:
+ builder.append('\t' as char)
+ break
+ case 'u' as char:
+ if (position + 4 <= text.length()) {
+ String hex = text.substring(position, position
+ 4)
+ if
(UNICODE_ESCAPE_PATTERN.matcher(hex).matches()) {
+ builder.append((char)
Integer.parseInt(hex, 16))
+ position += 4
+ break
+ }
+ }
+ builder.append(escaped)
+ break
+ default:
+ builder.append(escaped)
+ break
+ }
+ } else {
+ builder.append(ch)
+ }
+ }
+ throw new IllegalArgumentException("Unclosed quoted string")
+ }
+
+ private Object parseToken(boolean stopAtColon) {
+ int start = position
+ while (position < text.length()) {
+ char ch = text.charAt(position)
+ if (ch == ',' as char || ch == ']' as char || ch == '}' as char
+ || (stopAtColon && ch == ':' as char)) {
+ break
+ }
+ position++
+ }
+ String token = text.substring(start, position).trim()
+ if (token.isEmpty()) {
+ return ""
+ }
+ String lowerToken = token.toLowerCase(Locale.ROOT)
+ if (lowerToken == "null") {
+ return null
+ }
+ if (lowerToken == "true") {
+ return true
+ }
+ if (lowerToken == "false") {
+ return false
+ }
+ if (NUMBER_PATTERN.matcher(token).matches()) {
+ return normalizeDecimal(new BigDecimal(token))
+ }
+ return token
+ }
+
+ private void skipWhitespace() {
+ while (position < text.length() &&
Character.isWhitespace(text.charAt(position))) {
+ position++
+ }
+ }
+
+ private boolean consume(char expected) {
+ if (position < text.length() && text.charAt(position) == expected)
{
+ position++
+ return true
+ }
+ return false
+ }
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_spark_doris_consistency_demo.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_spark_doris_consistency_demo.groovy
new file mode 100644
index 00000000000..3ce5c8ebef3
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_spark_doris_consistency_demo.groovy
@@ -0,0 +1,301 @@
+// 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.
+
+suite("test_iceberg_spark_doris_consistency_demo", "p0,external,iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable iceberg test.")
+ return
+ }
+
+ String catalogName = "test_iceberg_spark_doris_consistency_demo"
+ String dbName = "iceberg_spark_doris_consistency_demo_db"
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+
+ def expectedBasicRows = [
+ [1, "alice", 10],
+ [2, "bob", 20],
+ [3, "cindy", null],
+ [4, "doris", 40],
+ [5, "edge", 0]
+ ]
+ def expectedAggRows = [[5L, 70L]]
+
+ // Example: execute multiple Spark Iceberg statements in one JDBC
connection.
+ spark_iceberg_multi """
+ SET spark.sql.binaryOutputStyle=HEX;
+ SET spark.sql.timestampType=TIMESTAMP_NTZ;
+ CREATE DATABASE IF NOT EXISTS demo.${dbName};
+ DROP TABLE IF EXISTS demo.${dbName}.spark_written_iceberg_demo;
+ CREATE TABLE demo.${dbName}.spark_written_iceberg_demo (
+ id INT,
+ name STRING,
+ score INT,
+ string_col STRING,
+ bool_col BOOLEAN,
+ int_col INT,
+ bigint_col BIGINT,
+ float_col FLOAT,
+ double_col DOUBLE,
+ decimal_small_col DECIMAL(9, 2),
+ decimal_col DECIMAL(18, 6),
+ decimal_wide_col DECIMAL(38, 12),
+ date_col DATE,
+ timestamp_col TIMESTAMP,
+ binary_col BINARY,
+ array_col ARRAY<INT>,
+ array_string_col ARRAY<STRING>,
+ array_bool_col ARRAY<BOOLEAN>,
+ array_binary_col ARRAY<BINARY>,
+ array_decimal_col ARRAY<DECIMAL(18, 6)>,
+ array_date_col ARRAY<DATE>,
+ array_timestamp_col ARRAY<TIMESTAMP>,
+ map_col MAP<STRING, INT>,
+ map_int_string_col MAP<INT, STRING>,
+ map_bool_col MAP<BOOLEAN, BOOLEAN>,
+ map_binary_col MAP<STRING, BINARY>,
+ map_decimal_col MAP<DECIMAL(8, 2), DECIMAL(8, 2)>,
+ struct_col STRUCT<city:STRING, zip:INT>,
+ struct_all_col STRUCT<string_field:STRING, bool_field:BOOLEAN,
int_field:INT,
+ bigint_field:BIGINT, float_field:FLOAT,
double_field:DOUBLE,
+ binary_field:BINARY,
decimal_field:DECIMAL(18, 6), date_field:DATE,
+ timestamp_field:TIMESTAMP>,
+ nested_col MAP<STRING, ARRAY<STRUCT<score:INT, label:STRING>>>
+ ) USING iceberg;
+ INSERT INTO demo.${dbName}.spark_written_iceberg_demo VALUES
+ (
+ 1, 'alice', 10, 'alice-string', true,
+ 700, 7000000000,
+ CAST(1.25 AS FLOAT), CAST(10.125 AS DOUBLE),
+ CAST(12.34 AS DECIMAL(9, 2)),
+ CAST(12345.678900 AS DECIMAL(18, 6)),
+ CAST(123456789.012345678901 AS DECIMAL(38, 12)),
+ DATE '2024-03-20', TIMESTAMP '2024-03-20 12:00:00.123456',
+ CAST('alice-bin' AS BINARY),
+ ARRAY(1, 2, 3), ARRAY(CAST(NULL AS STRING), 'a', 'b'),
+ ARRAY(true, false),
+ ARRAY(CAST('alice-array-bin' AS BINARY), CAST(NULL AS BINARY)),
+ ARRAY(CAST(1.250000 AS DECIMAL(18, 6)), CAST(2.500000 AS
DECIMAL(18, 6))),
+ ARRAY(DATE '2024-03-20', DATE '2024-03-21'),
+ ARRAY(TIMESTAMP '2024-03-20 12:00:00.123456'),
+ MAP('math', 90, 'eng', 95),
+ MAP(1, 'one', 2, 'two'),
+ MAP(true, false, false, true),
+ MAP('payload', CAST('alice-map-bin' AS BINARY)),
+ MAP(CAST(1.25 AS DECIMAL(8, 2)), CAST(2.50 AS DECIMAL(8, 2)),
+ CAST(3.75 AS DECIMAL(8, 2)), CAST(4.00 AS DECIMAL(8, 2))),
+ NAMED_STRUCT('city', 'Beijing', 'zip', 100000),
+ NAMED_STRUCT('string_field', 'alice', 'bool_field', true,
'int_field', 700,
+ 'bigint_field', 7000000000, 'float_field', CAST(1.25 AS
FLOAT),
+ 'double_field', CAST(10.125 AS DOUBLE),
+ 'binary_field', CAST('alice-struct-bin' AS BINARY),
+ 'decimal_field', CAST(12345.678900 AS DECIMAL(18, 6)),
+ 'date_field', DATE '2024-03-20',
+ 'timestamp_field', TIMESTAMP '2024-03-20 12:00:00.123456'),
+ MAP('term', ARRAY(NAMED_STRUCT('score', 90, 'label', 'good')))
+ ),
+ (
+ 2, 'bob', 20, 'bob-string', false,
+ -800, -8000000000,
+ CAST(-2.5 AS FLOAT), CAST(-20.25 AS DOUBLE),
+ CAST(-98.76 AS DECIMAL(9, 2)),
+ CAST(-9876.543210 AS DECIMAL(18, 6)),
+ CAST(-987654321.012345678901 AS DECIMAL(38, 12)),
+ DATE '2024-03-21', TIMESTAMP '2024-03-21 13:01:02.654321',
+ CAST('bob-bin' AS BINARY),
+ ARRAY(4, 5), ARRAY('x', CAST(NULL AS STRING), 'z'),
+ ARRAY(false, true),
+ ARRAY(CAST('bob-array-bin' AS BINARY), CAST('bob-array-bin-2'
AS BINARY)),
+ ARRAY(CAST(-3.750000 AS DECIMAL(18, 6)), CAST(4.125000 AS
DECIMAL(18, 6))),
+ ARRAY(DATE '2024-03-21'),
+ ARRAY(TIMESTAMP '2024-03-21 13:01:02.654321',
+ TIMESTAMP '2024-03-21 13:01:03.000000'),
+ MAP('math', 80, 'eng', 85),
+ MAP(3, 'three', 4, 'four'),
+ MAP(true, true, false, false),
+ MAP('payload', CAST('bob-map-bin' AS BINARY)),
+ MAP(CAST(-1.25 AS DECIMAL(8, 2)), CAST(-2.50 AS DECIMAL(8,
2))),
+ NAMED_STRUCT('city', 'Shanghai', 'zip', 200000),
+ NAMED_STRUCT('string_field', 'bob', 'bool_field', false,
'int_field', -800,
+ 'bigint_field', -8000000000, 'float_field', CAST(-2.5 AS
FLOAT),
+ 'double_field', CAST(-20.25 AS DOUBLE),
+ 'binary_field', CAST('bob-struct-bin' AS BINARY),
+ 'decimal_field', CAST(-9876.543210 AS DECIMAL(18, 6)),
+ 'date_field', DATE '2024-03-21',
+ 'timestamp_field', TIMESTAMP '2024-03-21 13:01:02.654321'),
+ MAP('term', ARRAY(
+ NAMED_STRUCT('score', 80, 'label', 'pass'),
+ NAMED_STRUCT('score', 85, 'label', 'better')
+ ))
+ ),
+ (
+ 3, 'cindy', NULL, NULL, NULL,
+ NULL, NULL,
+ NULL, NULL, NULL, NULL, NULL,
+ NULL, NULL, NULL,
+ ARRAY(CAST(NULL AS INT), 6), ARRAY(CAST(NULL AS STRING)),
+ ARRAY(CAST(NULL AS BOOLEAN), true),
+ ARRAY(CAST(NULL AS BINARY)),
+ ARRAY(CAST(NULL AS DECIMAL(18, 6))),
+ ARRAY(CAST(NULL AS DATE)),
+ ARRAY(CAST(NULL AS TIMESTAMP)),
+ MAP('science', CAST(NULL AS INT)),
+ MAP(5, CAST(NULL AS STRING)),
+ MAP(false, CAST(NULL AS BOOLEAN)),
+ MAP('payload', CAST(NULL AS BINARY)),
+ MAP(CAST(5.25 AS DECIMAL(8, 2)), CAST(NULL AS DECIMAL(8, 2))),
+ NAMED_STRUCT('city', CAST(NULL AS STRING), 'zip', CAST(NULL AS
INT)),
+ NAMED_STRUCT('string_field', CAST(NULL AS STRING),
'bool_field', CAST(NULL AS BOOLEAN),
+ 'int_field', CAST(NULL AS INT),
+ 'bigint_field', CAST(NULL AS BIGINT), 'float_field',
CAST(NULL AS FLOAT),
+ 'double_field', CAST(NULL AS DOUBLE),
+ 'binary_field', CAST(NULL AS BINARY),
+ 'decimal_field', CAST(NULL AS DECIMAL(18, 6)),
+ 'date_field', CAST(NULL AS DATE),
+ 'timestamp_field', CAST(NULL AS TIMESTAMP)),
+ NULL
+ );
+ """
+
+ // Example: write one more Iceberg row through Spark SQL.
+ spark_iceberg """
+ INSERT INTO demo.${dbName}.spark_written_iceberg_demo VALUES
+ (
+ 4, 'doris', 40, 'doris-string', true,
+ 4000, 4000000000,
+ CAST(4.5 AS FLOAT), CAST(40.75 AS DOUBLE),
+ CAST(44.44 AS DECIMAL(9, 2)),
+ CAST(4444.000001 AS DECIMAL(18, 6)),
+ CAST(444444444.000000000001 AS DECIMAL(38, 12)),
+ DATE '2024-03-22', TIMESTAMP '2024-03-22 14:02:03.000001',
+ CAST('doris-bin' AS BINARY),
+ ARRAY(7, 8, 9), ARRAY('d', 'o', 'ris'),
+ ARRAY(true, true),
+ ARRAY(CAST('doris-array-bin' AS BINARY)),
+ ARRAY(CAST(4.000001 AS DECIMAL(18, 6))),
+ ARRAY(DATE '2024-03-22', DATE '2024-03-23'),
+ ARRAY(TIMESTAMP '2024-03-22 14:02:03.000001'),
+ MAP('math', 100, 'eng', 99),
+ MAP(6, 'six', 7, 'seven'),
+ MAP(true, false),
+ MAP('payload', CAST('doris-map-bin' AS BINARY)),
+ MAP(CAST(6.25 AS DECIMAL(8, 2)), CAST(7.50 AS DECIMAL(8, 2))),
+ NAMED_STRUCT('city', 'Chengdu', 'zip', 610000),
+ NAMED_STRUCT('string_field', 'doris', 'bool_field', true,
'int_field', 4000,
+ 'bigint_field', 4000000000, 'float_field', CAST(4.5 AS
FLOAT),
+ 'double_field', CAST(40.75 AS DOUBLE),
+ 'binary_field', CAST('doris-struct-bin' AS BINARY),
+ 'decimal_field', CAST(4444.000001 AS DECIMAL(18, 6)),
+ 'date_field', DATE '2024-03-22',
+ 'timestamp_field', TIMESTAMP '2024-03-22 14:02:03.000001'),
+ MAP('term', ARRAY(NAMED_STRUCT('score', 100, 'label',
'excellent')))
+ ),
+ (
+ 5, 'edge', 0, '', false,
+ 0, CAST('-9223372036854775808' AS BIGINT),
+ CAST(0.1 AS FLOAT), CAST(-0.1 AS DOUBLE),
+ CAST(0.00 AS DECIMAL(9, 2)),
+ CAST(-0.000001 AS DECIMAL(18, 6)),
+ CAST(99999999999999999999999999.999999999999 AS DECIMAL(38,
12)),
+ DATE '1970-01-01', TIMESTAMP '1970-01-01 00:00:00.000000',
+ CAST('' AS BINARY),
+ CAST(ARRAY() AS ARRAY<INT>),
+ ARRAY('', 'space value', 'edge-value'),
+ CAST(ARRAY() AS ARRAY<BOOLEAN>),
+ CAST(ARRAY() AS ARRAY<BINARY>),
+ CAST(ARRAY() AS ARRAY<DECIMAL(18, 6)>),
+ CAST(ARRAY() AS ARRAY<DATE>),
+ CAST(ARRAY() AS ARRAY<TIMESTAMP>),
+ map_from_arrays(CAST(ARRAY() AS ARRAY<STRING>), CAST(ARRAY()
AS ARRAY<INT>)),
+ map_from_arrays(CAST(ARRAY() AS ARRAY<INT>), CAST(ARRAY() AS
ARRAY<STRING>)),
+ map_from_arrays(CAST(ARRAY() AS ARRAY<BOOLEAN>), CAST(ARRAY()
AS ARRAY<BOOLEAN>)),
+ map_from_arrays(CAST(ARRAY() AS ARRAY<STRING>), CAST(ARRAY()
AS ARRAY<BINARY>)),
+ map_from_arrays(CAST(ARRAY() AS ARRAY<DECIMAL(8, 2)>),
+ CAST(ARRAY() AS ARRAY<DECIMAL(8, 2)>)),
+ NAMED_STRUCT('city', '', 'zip', 0),
+ NAMED_STRUCT('string_field', '', 'bool_field', false,
'int_field', 0,
+ 'bigint_field', CAST('-9223372036854775808' AS BIGINT),
+ 'float_field', CAST(-0.0 AS FLOAT), 'double_field',
CAST(0.0 AS DOUBLE),
+ 'binary_field', CAST('' AS BINARY),
+ 'decimal_field', CAST(-0.000001 AS DECIMAL(18, 6)),
+ 'date_field', DATE '1970-01-01',
+ 'timestamp_field', TIMESTAMP '1970-01-01 00:00:00.000000'),
+ MAP('empty', CAST(ARRAY() AS ARRAY<STRUCT<score:INT,
label:STRING>>),
+ 'blank', ARRAY(NAMED_STRUCT('score', 0, 'label', '')))
+ );
+ """
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type'='iceberg',
+ 'iceberg.catalog.type'='rest',
+ 'uri' = 'http://${externalEnvIp}:${restPort}',
+ 's3.access_key' = 'admin',
+ 's3.secret_key' = 'password',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
+ 's3.region' = 'us-east-1',
+ 'enable.mapping.varbinary' = 'true'
+ );
+ """
+
+ sql """switch ${catalogName}"""
+
+ def sparkBasicRows = spark_iceberg """
+ SELECT id, name, score
+ FROM demo.${dbName}.spark_written_iceberg_demo
+ ORDER BY id
+ """
+ // Example 1: compare Spark Iceberg query result with explicit expected
values.
+ assertEquals(expectedBasicRows, sparkBasicRows)
+
+ def dorisBasicRows = sql """
+ SELECT id, name, score
+ FROM ${dbName}.spark_written_iceberg_demo
+ ORDER BY id
+ """
+ // Example 1: compare Doris Iceberg query result with explicit expected
values.
+ assertEquals(expectedBasicRows, dorisBasicRows)
+
+ // Example 2: compare Doris and Spark query results.
+ def sparkRows = spark_iceberg """
+ SELECT *
+ FROM demo.${dbName}.spark_written_iceberg_demo
+ ORDER BY id
+ """
+ def dorisRows = sql """
+ SELECT *
+ FROM ${dbName}.spark_written_iceberg_demo
+ ORDER BY id
+ """
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+
+ def sparkAggRows = spark_iceberg """
+ SELECT count(*), sum(score)
+ FROM demo.${dbName}.spark_written_iceberg_demo
+ """
+ // Compare Spark Iceberg aggregate result with explicit expected values.
+ assertEquals(expectedAggRows, sparkAggRows)
+
+ def dorisAggRows = sql """
+ SELECT count(*), sum(score)
+ FROM ${dbName}.spark_written_iceberg_demo
+ """
+ assertSparkDorisResultEquals(sparkAggRows, dorisAggRows)
+}
diff --git
a/regression-test/suites/external_table_p0/paimon/test_paimon_spark_doris_consistency_demo.groovy
b/regression-test/suites/external_table_p0/paimon/test_paimon_spark_doris_consistency_demo.groovy
new file mode 100644
index 00000000000..ab2cfe4090c
--- /dev/null
+++
b/regression-test/suites/external_table_p0/paimon/test_paimon_spark_doris_consistency_demo.groovy
@@ -0,0 +1,317 @@
+// 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.
+
+suite("test_paimon_spark_doris_consistency_demo", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test.")
+ return
+ }
+
+ String catalogName = "test_paimon_spark_doris_consistency_demo"
+ String dbName = "paimon_spark_doris_consistency_demo_db"
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+
+ def expectedBasicRows = [
+ [1, "alice", 10],
+ [2, "bob", 20],
+ [3, "cindy", null],
+ [4, "doris", 40],
+ [5, "edge", 0]
+ ]
+ def expectedAggRows = [[5L, 70L]]
+
+ // Example: execute multiple Spark Paimon statements in one JDBC
connection.
+ spark_paimon_multi """
+ SET spark.sql.binaryOutputStyle=HEX;
+ SET spark.sql.preserveCharVarcharTypeInfo=true;
+ SET spark.sql.timestampType=TIMESTAMP_NTZ;
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+ DROP TABLE IF EXISTS paimon.${dbName}.spark_written_paimon_demo;
+ CREATE TABLE paimon.${dbName}.spark_written_paimon_demo (
+ id INT,
+ name STRING,
+ score INT,
+ string_col STRING,
+ varchar_col VARCHAR(20),
+ char_col CHAR(10),
+ bool_col BOOLEAN,
+ tinyint_col TINYINT,
+ smallint_col SMALLINT,
+ int_col INT,
+ bigint_col BIGINT,
+ float_col FLOAT,
+ double_col DOUBLE,
+ decimal_small_col DECIMAL(9, 2),
+ decimal_col DECIMAL(18, 6),
+ decimal_wide_col DECIMAL(38, 12),
+ date_col DATE,
+ timestamp_col TIMESTAMP,
+ binary_col BINARY,
+ array_col ARRAY<INT>,
+ array_tinyint_col ARRAY<TINYINT>,
+ array_smallint_col ARRAY<SMALLINT>,
+ array_string_col ARRAY<STRING>,
+ array_bool_col ARRAY<BOOLEAN>,
+ array_binary_col ARRAY<BINARY>,
+ array_decimal_col ARRAY<DECIMAL(18, 6)>,
+ array_date_col ARRAY<DATE>,
+ array_timestamp_col ARRAY<TIMESTAMP>,
+ map_col MAP<STRING, INT>,
+ map_int_string_col MAP<INT, STRING>,
+ map_bool_col MAP<BOOLEAN, BOOLEAN>,
+ map_binary_col MAP<STRING, BINARY>,
+ map_decimal_col MAP<DECIMAL(8, 2), DECIMAL(8, 2)>,
+ struct_col STRUCT<city:STRING, zip:INT>,
+ struct_all_col STRUCT<string_field:STRING, bool_field:BOOLEAN,
int_field:INT,
+ bigint_field:BIGINT, float_field:FLOAT,
double_field:DOUBLE,
+ binary_field:BINARY,
decimal_field:DECIMAL(18, 6), date_field:DATE,
+ timestamp_field:TIMESTAMP>,
+ nested_col MAP<STRING, ARRAY<STRUCT<score:INT, label:STRING>>>
+ ) USING paimon;
+ INSERT INTO paimon.${dbName}.spark_written_paimon_demo VALUES
+ (
+ 1, 'alice', 10, 'alice-string', 'alice-varchar', 'alice_char',
true,
+ CAST(7 AS TINYINT), CAST(70 AS SMALLINT), 700, 7000000000,
+ CAST(1.25 AS FLOAT), CAST(10.125 AS DOUBLE),
+ CAST(12.34 AS DECIMAL(9, 2)),
+ CAST(12345.678900 AS DECIMAL(18, 6)),
+ CAST(123456789.012345678901 AS DECIMAL(38, 12)),
+ DATE '2024-03-20', TIMESTAMP '2024-03-20 12:00:00.123456',
+ CAST('alice-bin' AS BINARY),
+ ARRAY(1, 2, 3), ARRAY(CAST(1 AS TINYINT), CAST(2 AS TINYINT)),
+ ARRAY(CAST(10 AS SMALLINT), CAST(20 AS SMALLINT)),
+ ARRAY(CAST(NULL AS STRING), 'a', 'b'),
+ ARRAY(true, false),
+ ARRAY(CAST('alice-array-bin' AS BINARY), CAST(NULL AS BINARY)),
+ ARRAY(CAST(1.250000 AS DECIMAL(18, 6)), CAST(2.500000 AS
DECIMAL(18, 6))),
+ ARRAY(DATE '2024-03-20', DATE '2024-03-21'),
+ ARRAY(TIMESTAMP '2024-03-20 12:00:00.123456'),
+ MAP('math', 90, 'eng', 95),
+ MAP(1, 'one', 2, 'two'),
+ MAP(true, false, false, true),
+ MAP('payload', CAST('alice-map-bin' AS BINARY)),
+ MAP(CAST(1.25 AS DECIMAL(8, 2)), CAST(2.50 AS DECIMAL(8, 2)),
+ CAST(3.75 AS DECIMAL(8, 2)), CAST(4.00 AS DECIMAL(8, 2))),
+ NAMED_STRUCT('city', 'Beijing', 'zip', 100000),
+ NAMED_STRUCT('string_field', 'alice', 'bool_field', true,
'int_field', 700,
+ 'bigint_field', 7000000000, 'float_field', CAST(1.25 AS
FLOAT),
+ 'double_field', CAST(10.125 AS DOUBLE),
+ 'binary_field', CAST('alice-struct-bin' AS BINARY),
+ 'decimal_field', CAST(12345.678900 AS DECIMAL(18, 6)),
+ 'date_field', DATE '2024-03-20',
+ 'timestamp_field', TIMESTAMP '2024-03-20 12:00:00.123456'),
+ MAP('term', ARRAY(NAMED_STRUCT('score', 90, 'label', 'good')))
+ ),
+ (
+ 2, 'bob', 20, 'bob-string', 'bob-varchar', 'bob_char__', false,
+ CAST(-8 AS TINYINT), CAST(-80 AS SMALLINT), -800, -8000000000,
+ CAST(-2.5 AS FLOAT), CAST(-20.25 AS DOUBLE),
+ CAST(-98.76 AS DECIMAL(9, 2)),
+ CAST(-9876.543210 AS DECIMAL(18, 6)),
+ CAST(-987654321.012345678901 AS DECIMAL(38, 12)),
+ DATE '2024-03-21', TIMESTAMP '2024-03-21 13:01:02.654321',
+ CAST('bob-bin' AS BINARY),
+ ARRAY(4, 5), ARRAY(CAST(-1 AS TINYINT), CAST(-2 AS TINYINT)),
+ ARRAY(CAST(-10 AS SMALLINT), CAST(-20 AS SMALLINT)),
+ ARRAY('x', CAST(NULL AS STRING), 'z'),
+ ARRAY(false, true),
+ ARRAY(CAST('bob-array-bin' AS BINARY), CAST('bob-array-bin-2'
AS BINARY)),
+ ARRAY(CAST(-3.750000 AS DECIMAL(18, 6)), CAST(4.125000 AS
DECIMAL(18, 6))),
+ ARRAY(DATE '2024-03-21'),
+ ARRAY(TIMESTAMP '2024-03-21 13:01:02.654321',
+ TIMESTAMP '2024-03-21 13:01:03.000000'),
+ MAP('math', 80, 'eng', 85),
+ MAP(3, 'three', 4, 'four'),
+ MAP(true, true, false, false),
+ MAP('payload', CAST('bob-map-bin' AS BINARY)),
+ MAP(CAST(-1.25 AS DECIMAL(8, 2)), CAST(-2.50 AS DECIMAL(8,
2))),
+ NAMED_STRUCT('city', 'Shanghai', 'zip', 200000),
+ NAMED_STRUCT('string_field', 'bob', 'bool_field', false,
'int_field', -800,
+ 'bigint_field', -8000000000, 'float_field', CAST(-2.5 AS
FLOAT),
+ 'double_field', CAST(-20.25 AS DOUBLE),
+ 'binary_field', CAST('bob-struct-bin' AS BINARY),
+ 'decimal_field', CAST(-9876.543210 AS DECIMAL(18, 6)),
+ 'date_field', DATE '2024-03-21',
+ 'timestamp_field', TIMESTAMP '2024-03-21 13:01:02.654321'),
+ MAP('term', ARRAY(
+ NAMED_STRUCT('score', 80, 'label', 'pass'),
+ NAMED_STRUCT('score', 85, 'label', 'better')
+ ))
+ ),
+ (
+ 3, 'cindy', NULL, NULL, NULL, NULL, NULL,
+ NULL, NULL, NULL,
+ NULL, NULL, NULL, NULL, NULL,
+ NULL, NULL, NULL, NULL,
+ ARRAY(CAST(NULL AS INT), 6), ARRAY(CAST(NULL AS TINYINT)),
+ ARRAY(CAST(NULL AS SMALLINT)),
+ ARRAY(CAST(NULL AS STRING)),
+ ARRAY(CAST(NULL AS BOOLEAN), true),
+ ARRAY(CAST(NULL AS BINARY)),
+ ARRAY(CAST(NULL AS DECIMAL(18, 6))),
+ ARRAY(CAST(NULL AS DATE)),
+ ARRAY(CAST(NULL AS TIMESTAMP)),
+ MAP('science', CAST(NULL AS INT)),
+ MAP(5, CAST(NULL AS STRING)),
+ MAP(false, CAST(NULL AS BOOLEAN)),
+ MAP('payload', CAST(NULL AS BINARY)),
+ MAP(CAST(5.25 AS DECIMAL(8, 2)), CAST(NULL AS DECIMAL(8, 2))),
+ NAMED_STRUCT('city', CAST(NULL AS STRING), 'zip', CAST(NULL AS
INT)),
+ NAMED_STRUCT('string_field', CAST(NULL AS STRING),
'bool_field', CAST(NULL AS BOOLEAN),
+ 'int_field', CAST(NULL AS INT),
+ 'bigint_field', CAST(NULL AS BIGINT), 'float_field',
CAST(NULL AS FLOAT),
+ 'double_field', CAST(NULL AS DOUBLE),
+ 'binary_field', CAST(NULL AS BINARY),
+ 'decimal_field', CAST(NULL AS DECIMAL(18, 6)),
+ 'date_field', CAST(NULL AS DATE),
+ 'timestamp_field', CAST(NULL AS TIMESTAMP)),
+ NULL
+ );
+ """
+
+ // Example: write one more Paimon row through Spark SQL.
+ spark_paimon """
+ INSERT INTO paimon.${dbName}.spark_written_paimon_demo VALUES
+ (
+ 4, 'doris', 40, 'doris-string', 'doris-varchar', 'doris_char',
true,
+ CAST(4 AS TINYINT), CAST(400 AS SMALLINT), 4000, 4000000000,
+ CAST(4.5 AS FLOAT), CAST(40.75 AS DOUBLE),
+ CAST(44.44 AS DECIMAL(9, 2)),
+ CAST(4444.000001 AS DECIMAL(18, 6)),
+ CAST(444444444.000000000001 AS DECIMAL(38, 12)),
+ DATE '2024-03-22', TIMESTAMP '2024-03-22 14:02:03.000001',
+ CAST('doris-bin' AS BINARY),
+ ARRAY(7, 8, 9), ARRAY(CAST(3 AS TINYINT), CAST(4 AS TINYINT)),
+ ARRAY(CAST(30 AS SMALLINT), CAST(40 AS SMALLINT)),
+ ARRAY('d', 'o', 'ris'),
+ ARRAY(true, true),
+ ARRAY(CAST('doris-array-bin' AS BINARY)),
+ ARRAY(CAST(4.000001 AS DECIMAL(18, 6))),
+ ARRAY(DATE '2024-03-22', DATE '2024-03-23'),
+ ARRAY(TIMESTAMP '2024-03-22 14:02:03.000001'),
+ MAP('math', 100, 'eng', 99),
+ MAP(6, 'six', 7, 'seven'),
+ MAP(true, false),
+ MAP('payload', CAST('doris-map-bin' AS BINARY)),
+ MAP(CAST(6.25 AS DECIMAL(8, 2)), CAST(7.50 AS DECIMAL(8, 2))),
+ NAMED_STRUCT('city', 'Chengdu', 'zip', 610000),
+ NAMED_STRUCT('string_field', 'doris', 'bool_field', true,
'int_field', 4000,
+ 'bigint_field', 4000000000, 'float_field', CAST(4.5 AS
FLOAT),
+ 'double_field', CAST(40.75 AS DOUBLE),
+ 'binary_field', CAST('doris-struct-bin' AS BINARY),
+ 'decimal_field', CAST(4444.000001 AS DECIMAL(18, 6)),
+ 'date_field', DATE '2024-03-22',
+ 'timestamp_field', TIMESTAMP '2024-03-22 14:02:03.000001'),
+ MAP('term', ARRAY(NAMED_STRUCT('score', 100, 'label',
'excellent')))
+ ),
+ (
+ 5, 'edge', 0, '', '', 'edge_char_', false,
+ CAST(-128 AS TINYINT), CAST(-32768 AS SMALLINT), 0,
+ CAST('-9223372036854775808' AS BIGINT),
+ CAST(0.1 AS FLOAT), CAST(-0.1 AS DOUBLE),
+ CAST(0.00 AS DECIMAL(9, 2)),
+ CAST(-0.000001 AS DECIMAL(18, 6)),
+ CAST(99999999999999999999999999.999999999999 AS DECIMAL(38,
12)),
+ DATE '1970-01-01', TIMESTAMP '1970-01-01 00:00:00.000000',
+ CAST('' AS BINARY),
+ CAST(ARRAY() AS ARRAY<INT>),
+ CAST(ARRAY() AS ARRAY<TINYINT>),
+ CAST(ARRAY() AS ARRAY<SMALLINT>),
+ ARRAY('', 'space value', 'edge-value'),
+ CAST(ARRAY() AS ARRAY<BOOLEAN>),
+ CAST(ARRAY() AS ARRAY<BINARY>),
+ CAST(ARRAY() AS ARRAY<DECIMAL(18, 6)>),
+ CAST(ARRAY() AS ARRAY<DATE>),
+ CAST(ARRAY() AS ARRAY<TIMESTAMP>),
+ map_from_arrays(CAST(ARRAY() AS ARRAY<STRING>), CAST(ARRAY()
AS ARRAY<INT>)),
+ map_from_arrays(CAST(ARRAY() AS ARRAY<INT>), CAST(ARRAY() AS
ARRAY<STRING>)),
+ map_from_arrays(CAST(ARRAY() AS ARRAY<BOOLEAN>), CAST(ARRAY()
AS ARRAY<BOOLEAN>)),
+ map_from_arrays(CAST(ARRAY() AS ARRAY<STRING>), CAST(ARRAY()
AS ARRAY<BINARY>)),
+ map_from_arrays(CAST(ARRAY() AS ARRAY<DECIMAL(8, 2)>),
+ CAST(ARRAY() AS ARRAY<DECIMAL(8, 2)>)),
+ NAMED_STRUCT('city', '', 'zip', 0),
+ NAMED_STRUCT('string_field', '', 'bool_field', false,
'int_field', 0,
+ 'bigint_field', CAST('-9223372036854775808' AS BIGINT),
+ 'float_field', CAST(-0.0 AS FLOAT), 'double_field',
CAST(0.0 AS DOUBLE),
+ 'binary_field', CAST('' AS BINARY),
+ 'decimal_field', CAST(-0.000001 AS DECIMAL(18, 6)),
+ 'date_field', DATE '1970-01-01',
+ 'timestamp_field', TIMESTAMP '1970-01-01 00:00:00.000000'),
+ MAP('empty', CAST(ARRAY() AS ARRAY<STRUCT<score:INT,
label:STRING>>),
+ 'blank', ARRAY(NAMED_STRUCT('score', 0, 'label', '')))
+ );
+ """
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type' = 'paimon',
+ 'warehouse' = 's3://warehouse/wh',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
+ 's3.access_key' = 'admin',
+ 's3.secret_key' = 'password',
+ 's3.path.style.access' = 'true',
+ 'enable.mapping.varbinary' = 'true'
+ );
+ """
+
+ sql """switch ${catalogName}"""
+
+ def sparkBasicRows = spark_paimon """
+ SELECT id, name, score
+ FROM paimon.${dbName}.spark_written_paimon_demo
+ ORDER BY id
+ """
+ // Example 1: compare Spark Paimon query result with explicit expected
values.
+ assertEquals(expectedBasicRows, sparkBasicRows)
+
+ def dorisBasicRows = sql """
+ SELECT id, name, score
+ FROM ${dbName}.spark_written_paimon_demo
+ ORDER BY id
+ """
+ // Example 1: compare Doris Paimon query result with explicit expected
values.
+ assertEquals(expectedBasicRows, dorisBasicRows)
+
+ // Example 2: compare Doris and Spark query results.
+ def sparkRows = spark_paimon """
+ SELECT *
+ FROM paimon.${dbName}.spark_written_paimon_demo
+ ORDER BY id
+ """
+ def dorisRows = sql """
+ SELECT *
+ FROM ${dbName}.spark_written_paimon_demo
+ ORDER BY id
+ """
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+
+ def sparkAggRows = spark_paimon """
+ SELECT count(*), sum(score)
+ FROM paimon.${dbName}.spark_written_paimon_demo
+ """
+ // Compare Spark Paimon aggregate result with explicit expected values.
+ assertEquals(expectedAggRows, sparkAggRows)
+
+ def dorisAggRows = sql """
+ SELECT count(*), sum(score)
+ FROM ${dbName}.spark_written_paimon_demo
+ """
+ assertSparkDorisResultEquals(sparkAggRows, dorisAggRows)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
