This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 0d40ac56c6d branch-4.1: [improvement](regression) Use Spark thrift
JDBC for exter… (#65198)
0d40ac56c6d is described below
commit 0d40ac56c6de9f6f9e937e3e4730ff6735df4e0e
Author: zgxme <[email protected]>
AuthorDate: Sat Jul 4 12:31:35 2026 +0800
branch-4.1: [improvement](regression) Use Spark thrift JDBC for exter…
(#65198)
### What problem does this PR solve?
bp https://github.com/apache/doris/pull/64886
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../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 | 133 ++--
.../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, 1505 insertions(+), 91 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 83c1ee6d031..e4e922d1666 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 192f1c2fb7f..0075e501545 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
@@ -1608,74 +1609,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() {
- 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()
-
- if (output) {
- // Get the first matching container
- String containerName = output.split('\n')[0].trim()
- if (containerName) {
- logger.info("Found spark-iceberg container:
${containerName}".toString())
- return containerName
- }
- }
+ 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)
+ }
- 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
+ private List spark_sql_multi(Object sqlStatements, boolean isOrder =
false) {
+ def statements = sqlStatements.toString().split(';').collect {
it.trim() }.findAll { it }
+
+ 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 on one JDBC connection to reduce startup
overhead.
*
* Usage:
* spark_iceberg_multi '''
@@ -1684,49 +1652,40 @@ class Suite implements GroovyInterceptable {
* INSERT INTO demo.test_db.t1 VALUES (1);
* '''
*/
- List<String> spark_iceberg_multi(String sqlStatements, int timeoutSeconds
= 300) {
- // Split by semicolon and execute each statement
- def statements = sqlStatements.split(';').collect { it.trim()
}.findAll { it }
- def results = []
-
- for (stmt in statements) {
- if (stmt) {
- results << spark_iceberg(stmt, timeoutSeconds)
- }
- }
-
- return results
+ 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) {
@@ -1950,6 +1909,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 1fbdc2f45fc..dd360b353b8 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
@@ -235,6 +236,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) {
@@ -310,6 +320,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")
@@ -612,6 +637,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]
