github-actions[bot] commented on code in PR #65329:
URL: https://github.com/apache/doris/pull/65329#discussion_r3567827674
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java:
##########
@@ -894,18 +1016,96 @@ public void modifyColumn(ExternalTable dorisTable,
Column column, ColumnPosition
refreshTable(dorisTable, updateTime);
}
+ @Override
+ public void modifyColumn(ExternalTable dorisTable, ColumnPath columnPath,
Column column, ColumnPosition position,
+ long updateTime) throws UserException {
+ if (!columnPath.isNested()) {
+ modifyColumn(dorisTable, column, position, updateTime);
+ return;
+ }
+
+ Table icebergTable = IcebergUtils.getIcebergTable(dorisTable);
+ ResolvedColumnPath resolvedPath =
validateNestedStructFieldPath(icebergTable.schema(), columnPath, "modify");
Review Comment:
This rejects the direct array-element/map-value type changes described by
the PR. For `MODIFY COLUMN arr.element BIGINT` the nested path reaches this
call with parent path `arr`; `validateNestedStructFieldPath` resolves that
parent to a LIST and throws because it is not a struct. `m.value` fails the
same way with a MAP parent, so the later `applyListChange`/`applyMapChange`
logic is never reached. Please route direct `*.element` and `*.value` modify
targets through `resolveColumnPath` and the list/map update logic, or remove
the advertised syntax and add a negative test; the current tests only cover
child fields like `arr.element.x` and `m.value.v`.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java:
##########
@@ -833,12 +908,33 @@ public void dropColumn(ExternalTable dorisTable, String
columnName, long updateT
refreshTable(dorisTable, updateTime);
}
+ @Override
+ public void dropColumn(ExternalTable dorisTable, ColumnPath columnPath,
long updateTime) throws UserException {
+ if (!columnPath.isNested()) {
+ dropColumn(dorisTable, columnPath.getTopLevelName(), updateTime);
+ return;
+ }
+ Table icebergTable = IcebergUtils.getIcebergTable(dorisTable);
+ ResolvedColumnPath resolvedPath =
validateNestedStructFieldPath(icebergTable.schema(), columnPath, "drop");
+
+ UpdateSchema updateSchema = icebergTable.updateSchema();
+ updateSchema.deleteColumn(resolvedPath.getFullPath());
+ try {
+ executionAuthenticator.execute(() -> updateSchema.commit());
+ } catch (Exception e) {
+ throw new UserException("Failed to drop nested column: " +
columnPath.getFullPath() + " from table: "
+ + icebergTable.name() + ", error message is: " +
e.getMessage(), e);
+ }
+ refreshTable(dorisTable, updateTime);
+ }
+
@Override
public void renameColumn(ExternalTable dorisTable, String oldName, String
newName, long updateTime)
throws UserException {
Table icebergTable = IcebergUtils.getIcebergTable(dorisTable);
+ ResolvedColumnPath oldPath = resolveColumnPath(icebergTable.schema(),
ColumnPath.of(oldName), "rename");
UpdateSchema updateSchema = icebergTable.updateSchema();
- updateSchema.renameColumn(oldName, newName);
+ updateSchema.renameColumn(oldPath.getFullPath(), newName);
Review Comment:
The top-level path still needs the same case-insensitive collision guard
that was added for nested ADD/RENAME. Here we resolve the old name
case-insensitively, but then pass raw `newName` to Iceberg;
`addOneColumn`/`addColumns` likewise pass raw `column.getName()` to
`addColumn`. For a Spark-created schema with `Id` and another field, Doris can
now accept `ALTER TABLE t RENAME COLUMN label TO id` or `ADD COLUMN id ...`,
leaving Iceberg with exact-case-distinct fields that Doris later lower-cases to
the same visible column. Please check the root struct for `newName`/new column
names with the same case-insensitive sibling rule, allowing only same-field
case-only renames, and add top-level mixed-case coverage.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_nested_schema_evolution_spark_doris_interop.groovy:
##########
@@ -0,0 +1,341 @@
+// 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_nested_schema_evolution_spark_doris_interop",
"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_nested_schema_evolution_spark_doris_interop"
+ String dbName = "iceberg_nested_schema_evolution_interop_db"
+ String dorisTable = "doris_nested_evolution_to_spark"
+ String sparkTable = "spark_nested_evolution_to_doris"
+ String mixedCaseTable = "spark_mixed_case_nested_collision"
+ 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")
+
+ 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",
+ "meta.cache.iceberg.table.ttl-second" = "0"
+ );
+ """
+
+ try {
+ sql """switch ${catalogName}"""
+ sql """drop database if exists ${dbName} force"""
+ sql """create database ${dbName}"""
+ sql """use ${dbName}"""
+ sql """set enable_fallback_to_original_planner=false"""
+
+ spark_iceberg """CREATE DATABASE IF NOT EXISTS demo.${dbName}"""
+
+ sql """
+ CREATE TABLE ${dorisTable} (
+ id INT NOT NULL,
+ info STRUCT<metric:INT, label:STRING>,
+ events ARRAY<STRUCT<score:INT>>,
+ attrs MAP<STRING, STRUCT<code:INT>>
+ )
+ """
+ sql """
+ INSERT INTO ${dorisTable} VALUES (
+ 1,
+ STRUCT(10, 'doris_before'),
+ ARRAY(STRUCT(100)),
+ MAP('k', STRUCT(1000))
+ )
+ """
+
+ sql """ALTER TABLE ${dorisTable} ADD COLUMN info.doris_added STRING
NULL COMMENT 'added by doris'"""
+ sql """ALTER TABLE ${dorisTable} ADD COLUMN info.doris_first STRING
NULL FIRST"""
+ sql """ALTER TABLE ${dorisTable} ADD COLUMN info.doris_after_metric
STRING NULL AFTER metric"""
+ sql """ALTER TABLE ${dorisTable} ADD COLUMN events.element.doris_score
INT NULL"""
+ sql """ALTER TABLE ${dorisTable} ADD COLUMN
events.element.doris_first_score INT NULL FIRST"""
+ sql """ALTER TABLE ${dorisTable} ADD COLUMN
events.element.doris_after_score INT NULL AFTER score"""
+ sql """ALTER TABLE ${dorisTable} ADD COLUMN attrs.value.doris_code INT
NULL"""
+ sql """ALTER TABLE ${dorisTable} ADD COLUMN
attrs.value.doris_first_code INT NULL FIRST"""
+ sql """ALTER TABLE ${dorisTable} ADD COLUMN
attrs.value.doris_after_code INT NULL AFTER code"""
+ sql """ALTER TABLE ${dorisTable} ADD COLUMN info.drop_me STRING NULL"""
+ sql """ALTER TABLE ${dorisTable} ADD COLUMN events.element.drop_me INT
NULL"""
+ sql """ALTER TABLE ${dorisTable} ADD COLUMN attrs.value.drop_me INT
NULL"""
+ sql """ALTER TABLE ${dorisTable} MODIFY COLUMN info.metric BIGINT"""
+ sql """ALTER TABLE ${dorisTable} MODIFY COLUMN events.element.score
BIGINT"""
+ sql """ALTER TABLE ${dorisTable} MODIFY COLUMN attrs.value.code
BIGINT"""
+ sql """ALTER TABLE ${dorisTable} RENAME COLUMN info.doris_added TO
doris_renamed"""
+ sql """ALTER TABLE ${dorisTable} RENAME COLUMN
events.element.doris_score TO doris_score_renamed"""
+ sql """ALTER TABLE ${dorisTable} RENAME COLUMN attrs.value.doris_code
TO doris_code_renamed"""
+ sql """ALTER TABLE ${dorisTable} MODIFY COLUMN info.doris_renamed
COMMENT 'renamed by doris'"""
+ sql """ALTER TABLE ${dorisTable} MODIFY COLUMN
events.element.doris_score_renamed COMMENT 'renamed by doris'"""
+ sql """ALTER TABLE ${dorisTable} MODIFY COLUMN
attrs.value.doris_code_renamed COMMENT 'renamed by doris'"""
+ sql """ALTER TABLE ${dorisTable} DROP COLUMN info.drop_me"""
+ sql """ALTER TABLE ${dorisTable} DROP COLUMN events.element.drop_me"""
+ sql """ALTER TABLE ${dorisTable} DROP COLUMN attrs.value.drop_me"""
+
+ spark_iceberg """REFRESH TABLE demo.${dbName}.${dorisTable}"""
+ spark_iceberg """
+ INSERT INTO demo.${dbName}.${dorisTable} VALUES (
+ 2,
+ NAMED_STRUCT('doris_first', 'spark_first',
+ 'metric', CAST(20 AS BIGINT),
+ 'doris_after_metric', 'spark_after_metric',
+ 'label', 'spark_after_doris',
+ 'doris_renamed', 'spark_can_write_doris_field'),
+ ARRAY(NAMED_STRUCT('doris_first_score', 202,
+ 'score', CAST(200 AS BIGINT),
+ 'doris_after_score', 203,
+ 'doris_score_renamed', 201)),
+ MAP('k', NAMED_STRUCT('doris_first_code', 2002,
+ 'code', CAST(2000 AS BIGINT),
+ 'doris_after_code', 2003,
+ 'doris_code_renamed', 2001))
+ )
+ """
+
+ sql """refresh table ${dbName}.${dorisTable}"""
+ def dorisDrivenSparkRows = spark_iceberg """
+ SELECT id,
+ info.doris_first,
+ info.metric,
+ info.doris_after_metric,
+ info.label,
+ info.doris_renamed,
+ events[0].doris_first_score,
+ events[0].score,
+ events[0].doris_after_score,
+ events[0].doris_score_renamed,
+ attrs['k'].doris_first_code,
+ attrs['k'].code,
+ attrs['k'].doris_after_code,
+ attrs['k'].doris_code_renamed
+ FROM demo.${dbName}.${dorisTable}
+ ORDER BY id
+ """
+ def dorisDrivenDorisRows = sql """
+ SELECT id,
+ element_at(info, 'doris_first'),
+ element_at(info, 'metric'),
+ element_at(info, 'doris_after_metric'),
+ element_at(info, 'label'),
+ element_at(info, 'doris_renamed'),
+ element_at(events[1], 'doris_first_score'),
+ element_at(events[1], 'score'),
+ element_at(events[1], 'doris_after_score'),
+ element_at(events[1], 'doris_score_renamed'),
+ element_at(attrs['k'], 'doris_first_code'),
+ element_at(attrs['k'], 'code'),
+ element_at(attrs['k'], 'doris_after_code'),
+ element_at(attrs['k'], 'doris_code_renamed')
+ FROM ${dorisTable}
+ ORDER BY id
+ """
+ assertSparkDorisResultEquals(dorisDrivenSparkRows,
dorisDrivenDorisRows)
+
+ spark_iceberg_multi """
+ DROP TABLE IF EXISTS demo.${dbName}.${sparkTable};
+ DROP TABLE IF EXISTS demo.${dbName}.${mixedCaseTable};
+ CREATE TABLE demo.${dbName}.${mixedCaseTable} (
+ id INT,
+ Info STRUCT<Metric:INT, Label:STRING>
+ ) USING iceberg;
+ CREATE TABLE demo.${dbName}.${sparkTable} (
+ id INT,
+ info STRUCT<metric:INT, label:STRING>,
+ events ARRAY<STRUCT<score:INT>>,
+ attrs MAP<STRING, STRUCT<code:INT>>
+ ) USING iceberg;
+ INSERT INTO demo.${dbName}.${sparkTable} VALUES (
+ 1,
+ NAMED_STRUCT('metric', 10, 'label', 'spark_before'),
+ ARRAY(NAMED_STRUCT('score', 100)),
+ MAP('k', NAMED_STRUCT('code', 1000))
+ );
+ ALTER TABLE demo.${dbName}.${sparkTable} ADD COLUMN
info.spark_added STRING;
+ ALTER TABLE demo.${dbName}.${sparkTable} ADD COLUMN
info.spark_first STRING FIRST;
+ ALTER TABLE demo.${dbName}.${sparkTable} ADD COLUMN
info.spark_after_metric STRING AFTER metric;
+ ALTER TABLE demo.${dbName}.${sparkTable} ADD COLUMN
events.element.spark_score INT;
+ ALTER TABLE demo.${dbName}.${sparkTable} ADD COLUMN
events.element.spark_first_score INT FIRST;
+ ALTER TABLE demo.${dbName}.${sparkTable} ADD COLUMN
events.element.spark_after_score INT AFTER score;
+ ALTER TABLE demo.${dbName}.${sparkTable} ADD COLUMN
attrs.value.spark_code INT;
+ ALTER TABLE demo.${dbName}.${sparkTable} ADD COLUMN
attrs.value.spark_first_code INT FIRST;
+ ALTER TABLE demo.${dbName}.${sparkTable} ADD COLUMN
attrs.value.spark_after_code INT AFTER code;
+ ALTER TABLE demo.${dbName}.${sparkTable} ADD COLUMN info.drop_me
STRING;
+ ALTER TABLE demo.${dbName}.${sparkTable} ADD COLUMN
events.element.drop_me INT;
+ ALTER TABLE demo.${dbName}.${sparkTable} ADD COLUMN
attrs.value.drop_me INT;
+ ALTER TABLE demo.${dbName}.${sparkTable} ALTER COLUMN info.metric
TYPE BIGINT;
+ ALTER TABLE demo.${dbName}.${sparkTable} ALTER COLUMN
events.element.score TYPE BIGINT;
+ ALTER TABLE demo.${dbName}.${sparkTable} ALTER COLUMN
attrs.value.code TYPE BIGINT;
+ ALTER TABLE demo.${dbName}.${sparkTable} RENAME COLUMN
info.spark_added TO spark_renamed;
+ ALTER TABLE demo.${dbName}.${sparkTable} RENAME COLUMN
events.element.spark_score TO spark_score_renamed;
+ ALTER TABLE demo.${dbName}.${sparkTable} RENAME COLUMN
attrs.value.spark_code TO spark_code_renamed;
+ ALTER TABLE demo.${dbName}.${sparkTable} ALTER COLUMN
info.spark_renamed COMMENT 'renamed by spark';
+ ALTER TABLE demo.${dbName}.${sparkTable} ALTER COLUMN
events.element.spark_score_renamed COMMENT 'renamed by spark';
+ ALTER TABLE demo.${dbName}.${sparkTable} ALTER COLUMN
attrs.value.spark_code_renamed COMMENT 'renamed by spark';
+ ALTER TABLE demo.${dbName}.${sparkTable} DROP COLUMN info.drop_me;
+ ALTER TABLE demo.${dbName}.${sparkTable} DROP COLUMN
events.element.drop_me;
+ ALTER TABLE demo.${dbName}.${sparkTable} DROP COLUMN
attrs.value.drop_me;
+ INSERT INTO demo.${dbName}.${sparkTable} VALUES (
+ 2,
+ NAMED_STRUCT('spark_first', 'spark_first_field',
+ 'metric', CAST(20 AS BIGINT),
+ 'spark_after_metric', 'spark_after_metric_field',
+ 'label', 'spark_after',
+ 'spark_renamed', 'spark_new_field'),
+ ARRAY(NAMED_STRUCT('spark_first_score', 202,
+ 'score', CAST(200 AS BIGINT),
+ 'spark_after_score', 203,
+ 'spark_score_renamed', 201)),
+ MAP('k', NAMED_STRUCT('spark_first_code', 2002,
+ 'code', CAST(2000 AS BIGINT),
+ 'spark_after_code', 2003,
+ 'spark_code_renamed', 2001))
+ );
+ """
+
+ sql """refresh catalog ${catalogName}"""
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+
+ test {
+ sql """ALTER TABLE ${mixedCaseTable} ADD COLUMN info.metric STRING
NULL"""
+ exception "conflicts with existing Iceberg field 'Info.Metric'
(case-insensitive)"
+ }
+ test {
+ sql """ALTER TABLE ${mixedCaseTable} RENAME COLUMN info.label TO
metric"""
+ exception "conflicts with existing Iceberg field 'Info.Metric'
(case-insensitive)"
+ }
+
+ def sparkDrivenDescRows = sql """DESC ${sparkTable}"""
Review Comment:
This new suite still leaves deterministic Doris output outside the
regression baseline. `DESC ${sparkTable}` is checked with string `assertTrue`
calls here, and the ordered Doris result sets below are only compared in memory
against Spark rows; there is no generated
`test_iceberg_nested_schema_evolution_spark_doris_interop.out`. Please add
`qt_`/`order_qt_` checks for the deterministic Doris `DESC` and final query
outputs and include the generated `.out` file, while keeping the Spark/Doris
equality checks if they are useful for cross-engine parity.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
