hubgeter commented on code in PR #66620:
URL: https://github.com/apache/doris/pull/66620#discussion_r3754957410
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalIcebergMergeSink.java:
##########
@@ -73,9 +73,9 @@ public LogicalIcebergMergeSink(IcebergExternalDatabase
database,
this.targetIcebergTable = Objects.requireNonNull(
targetIcebergTable, "targetIcebergTable != null in
LogicalIcebergMergeSink");
// Delete-only MERGE writes position deletes and never invokes the
unsupported data writer.
- // UPDATE and data-producing MERGE must still reject an unchanged
Variant target column.
+ // UPDATE and data-producing MERGE must use the same Variant
capability checks as INSERT.
if (writesDataFiles) {
- IcebergUtils.validateWriteSchema(cols);
+ IcebergUtils.validateWriteSchema(targetIcebergTable, cols);
Review Comment:
Fixed in dbb637e5c87. Merge routing metadata is now identified by its fixed
output positions, so a legal target column named operation is handled as user
data and still receives target coercion. BindExpressionTest covers this case.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java:
##########
@@ -1407,6 +1421,24 @@ public static long getIcebergRowCount(ExternalTable tbl)
{
public static FileFormat getFileFormat(Table icebergTable) {
Map<String, String> properties = icebergTable.properties();
String fileFormatName = resolveFileFormatName(properties);
+ return parseFileFormatName(fileFormatName);
+ }
+
+ public static FileFormat getEffectiveFileFormat(Map<String, String>
tableProperties,
+ Map<String, String> catalogProperties) {
+ String fileFormatName =
catalogProperties.get(CatalogProperties.TABLE_OVERRIDE_PREFIX
Review Comment:
Fixed in dbb637e5c87. CREATE validation now resolves file format with the
same precedence as the runtime table properties: catalog defaults, table
properties, then catalog overrides. IcebergUtilsTest covers both precedence
directions.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java:
##########
@@ -726,12 +727,25 @@ public static boolean containsVariant(Type type) {
return false;
}
- public static void validateWriteSchema(List<Column> columns) {
- if (columns.stream().anyMatch(column ->
containsVariant(column.getType()))) {
- // Keep this table capability read-only until every Iceberg writer
can preserve the
- // Variant physical identity; rejecting only selected columns
would allow data loss.
+ public static void validateWriteSchema(Table table, List<Column> columns) {
+ if (columns.stream().noneMatch(column ->
containsVariant(column.getType()))) {
+ return;
+ }
+ validateWriteSchema(columns, getFormatVersion(table),
getFileFormat(table));
Review Comment:
Fixed in dbb637e5c87. Variant data-file writes now reject any query-eligible
smooth-upgrade source backend before planning. The backend compatibility cases
are covered in IcebergUtilsTest.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindExpression.java:
##########
@@ -297,31 +298,54 @@ private LogicalIcebergMergeSink<Plan>
bindIcebergMergeSink(
List<NamedExpression> outputExprs = sink.child().getOutput().stream()
.map(NamedExpression.class::cast)
.collect(ImmutableList.toImmutableList());
- List<Column> visibleColumns = sink.getCols().stream()
- .filter(Column::isVisible)
- .collect(ImmutableList.toImmutableList());
- int dataExprCount = 0;
- for (NamedExpression expr : outputExprs) {
- if (!isIcebergMergeMetaColumn(expr.getName())) {
- dataExprCount++;
- }
- }
- if (dataExprCount != visibleColumns.size()) {
+ List<NamedExpression> castExprs = coerceIcebergMergeOutput(
+ sink.getCols(), outputExprs, sink.isWritesDataFiles());
+ if (castExprs.equals(outputExprs)) {
if (sink.getOutputExprs().equals(outputExprs)) {
return sink;
}
return sink.withOutputExprs(outputExprs);
}
+ LogicalProject<?> project = new LogicalProject<>(castExprs,
sink.child());
+ return (LogicalIcebergMergeSink<Plan>)
sink.withChildAndUpdateOutput(project);
+ }
+
+ static List<NamedExpression> coerceIcebergMergeOutput(List<Column>
sinkColumns,
+ List<NamedExpression> outputExprs, boolean writesDataFiles) {
+ List<Column> outputColumns = sinkColumns.stream()
+ .filter(column -> column.isVisible() ||
IcebergUtils.isIcebergRowLineageColumn(column))
+ .collect(ImmutableList.toImmutableList());
+ int expectedOutputCount = 2 + outputColumns.size();
+ if (outputExprs.size() != expectedOutputCount
+ ||
!IcebergMergeOperation.OPERATION_COLUMN.equalsIgnoreCase(outputExprs.get(0).getName())
+ ||
!Column.ICEBERG_ROWID_COL.equalsIgnoreCase(outputExprs.get(1).getName())) {
+ throw new AnalysisException("Iceberg merge sink output is not
aligned with its routing metadata "
+ + "and target columns");
+ }
+
+ List<Column> visibleColumns =
Lists.newArrayListWithCapacity(outputColumns.size());
+ List<NamedExpression> visibleOutputExprs =
Lists.newArrayListWithCapacity(outputColumns.size());
+ for (int i = 0; i < outputColumns.size(); ++i) {
+ Column column = outputColumns.get(i);
+ if (column.isVisible()) {
+ visibleColumns.add(column);
+ visibleOutputExprs.add(outputExprs.get(i + 2));
+ }
+ }
+ if (writesDataFiles) {
+ IcebergVariantWriteAnalyzer.validate(visibleColumns,
visibleOutputExprs);
Review Comment:
Fixed in 132804a0b96. Each Variant-containing action projection is cast to
the target type before generateFinalProjections builds the nested IFs. Sink
validation then walks the resolved IF branches and inspects the pre-cast
source, preserving the legacy Variant rejection. Added unit coverage and an
end-to-end MERGE case that keeps an object Variant unchanged while inserting
primitive 1.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java:
##########
@@ -935,6 +965,7 @@ private void modifyTopLevelColumn(ExternalTable dorisTable,
ColumnPath columnPat
currentCol.type(), currentCol);
validateModifyColumnMetadata(column, resolvedPath.getFullPath(), true);
+ validateVariantSchema(icebergTable, column.getType(),
columnPath.getFullPath(), false);
Review Comment:
Fixed in 132804a0b96. Existing top-level Iceberg Variant columns now use a
dedicated metadata-only MODIFY path, support comment/nullability/position
changes, avoid asPrimitiveType, and explicitly reject conversions to or from
Variant. Added unit coverage plus CREATE/ADD followed by MODIFY and Spark-read
regression coverage.
--
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]
