sablejade opened a new issue, #9156: URL: https://github.com/apache/paimon/issues/9156
# [Bug][Spark] Positional dynamic partition writes can silently misalign columns ### Search before asking - [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar. ### Paimon version Apache Paimon master at commit <code>437bcc3c84a10b71b827a0a00c1661ccaad6e64f</code>. ### Compute Engine Apache Spark 3.4 with the Paimon Spark SQL Extension. ### Minimal reproduce step Build the <code>paimon-spark-3.4_2.12</code> bundle from the pinned commit, then start Spark SQL with that bundle and the following configuration: ~~~bash $SPARK_HOME/bin/spark-sql \ --jars /path/to/paimon-spark-3.4_2.12-2.1-SNAPSHOT.jar \ --conf spark.sql.extensions=org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions \ --conf spark.sql.sources.partitionOverwriteMode=dynamic \ --conf spark.sql.catalog.paimon=org.apache.paimon.spark.SparkCatalog \ --conf spark.sql.catalog.paimon.warehouse=file:///tmp/paimon-dynamic-partition-repro ~~~ The reproduction does not require <code>spark.paimon.write.use-v2-write=true</code>. The affected analyzer path is reachable with both the default value, <code>false</code>, and the explicit value, <code>true</code>. Create a Paimon namespace and a partitioned table: ~~~sql CREATE NAMESPACE IF NOT EXISTS paimon.repro; USE paimon.repro; CREATE TABLE target ( ds STRING, part STRING, uid STRING, key_name STRING, field_name STRING, value STRING, ttl STRING ) USING paimon PARTITIONED BY (ds, part); ~~~ Run this positional dynamic partition overwrite: ~~~sql INSERT OVERWRITE target PARTITION (ds, part) SELECT '20260810' AS ds, 'p1' AS part, 'user-1' AS uid, 'behavioral' AS key_name, 'metric-a' AS field_name, '0.5' AS detail_ratio, '3600' AS ttl UNION ALL SELECT '20260810' AS ds, 'p2' AS part, 'user-2' AS uid, 'behavioral' AS key_name, 'metric-b' AS field_name, '1' AS value, '3600' AS ttl; ~~~ Read the result: ~~~sql SELECT ds, part, uid, key_name, field_name, value, ttl FROM target ORDER BY ds, part, uid, key_name, field_name, value, ttl; ~~~ Expected rows: | ds | part | uid | key_name | field_name | value | ttl | |---|---|---|---|---|---|---| | 20260810 | p1 | user-1 | behavioral | metric-a | 0.5 | 3600 | | 20260810 | p2 | user-2 | behavioral | metric-b | 1 | 3600 | Actual rows on the affected implementation: | ds | part | uid | key_name | field_name | value | ttl | |---|---|---|---|---|---|---| | 0.5 | 3600 | 20260810 | p1 | user-1 | behavioral | metric-a | | 1 | 3600 | 20260810 | p2 | user-2 | behavioral | metric-b | ### What doesn't meet your expectations? A positional insert whose output is already in target-table order must not be interpreted as Hive-tail order only because one expression name differs from the target column name. The current behavior can silently corrupt data when the source and target types are compatible. The analyzer should preserve the table-ordered input in this case while continuing to support genuine Hive-style dynamic partition writes. ### Anything else? #### Root cause The affected marker and rewrite path was introduced by [#8414](https://github.com/apache/paimon/pull/8414) in commit [ce92c8a196](https://github.com/apache/paimon/commit/ce92c8a1969fab3d67c6440de6a96b43b945743e). It is included in Apache Paimon 2.0.0 and current master. The failure requires all of the following: 1. The insert has dynamic columns in <code>PARTITION (...)</code>, has no user-specified target column list, and is not a <code>BY NAME</code> write. 2. The query is already in target-table position order, but at least one output name differs from the corresponding target name. 3. The computed Hive-tail layout differs from the target table layout. If the dynamic partition columns are already at the end of the table schema, the existing second guard prevents the extra reorder. The reproduction deliberately declares <code>ds</code> and <code>part</code> first, so the two layouts are: ~~~text table: [ds, part, uid, key_name, field_name, value, ttl] Hive-tail: [uid, key_name, field_name, value, ttl, ds, part] ~~~ The parser rule <code>MarkHiveDynamicPartitionWrite</code> marks every matching insert without checking whether the query is already in table order: ~~~scala insert.userSpecifiedCols.isEmpty && !isByName(insert) && insert.partitionSpec.exists(_._2.isEmpty) ~~~ It wraps the query in <code>PaimonHiveDynamicPartitionQuery</code>. See [AbstractPaimonSparkSqlExtensionsParser.scala](https://github.com/apache/paimon/blob/437bcc3c84a10b71b827a0a00c1661ccaad6e64f/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala#L373-L386). Later, <code>PaimonAnalysis.resolveDynamicPartitionWrite</code> applies the Hive-style rewrite when both name checks fail: ~~~scala case Some(hiveStyleOutput) if !sameOutputNames(query.output, table.output) && !sameOutputNames(hiveStyleOutput, table.output) => ~~~ See [PaimonAnalysis.scala](https://github.com/apache/paimon/blob/437bcc3c84a10b71b827a0a00c1661ccaad6e64f/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala#L154-L179). In this reproduction: - <code>sameOutputNames(query.output, table.output)</code> is false because <code>detail_ratio != value</code>. - <code>sameOutputNames(hiveStyleOutput, table.output)</code> is false because <code>ds</code> and <code>part</code> are declared at the beginning of the table rather than at the end. The analyzer therefore treats the already table-ordered query as: ~~~text [uid, key_name, field_name, value, ttl, ds, part] ~~~ <details> <summary>Resulting column mapping</summary> The first positional resolution interprets the query as follows: | Query field | Interpreted as | |---|---| | ds | uid | | part | key_name | | uid | field_name | | key_name | value | | field_name | ttl | | detail_ratio | ds | | ttl | part | The subsequent by-name resolution produces: | Target field | Value taken from | |---|---| | ds | detail_ratio | | part | ttl | | uid | ds | | key_name | part | | field_name | uid | | value | key_name | | ttl | field_name | </details> #### Why UNION exposes the problem <code>UNION ALL</code> does not reorder the fields and does not create the Hive dynamic-partition marker. Its role is to preserve the correct position order while taking the output names from the first branch. The second branch alias <code>AS value</code> cannot change the UNION output name inherited from <code>AS detail_ratio</code> in the first branch. This makes the full-name table-order check fail. The resolved UNION output in the reproduction is therefore: ~~~text [ds, part, uid, key_name, field_name, detail_ratio, ttl] ~~~ UNION is not required. A single SELECT can trigger the same behavior if its output is already in table order but one output name differs. #### Affected write paths The parser marker can reach: - <code>AppendData</code>; - <code>OverwriteByExpression</code>; - <code>OverwritePartitionsDynamic</code>. With <code>spark.paimon.write.use-v2-write=false</code>, <code>OverwritePartitionsDynamic</code> is first converted to <code>PaimonDynamicPartitionOverwriteCommand</code>. That fallback command still implements <code>V2WriteCommand</code>, keeps the marked query as its child, and is matched by <code>PaimonAnalysis</code> again on a later analyzer pass. Therefore both values of <code>use-v2-write</code> are affected. See [PaimonDynamicPartitionOverwriteCommand.scala](https://github.com/apache/paimon/blob/437bcc3c84a10b71b827a0a00c1661ccaad6e64f/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonDynamicPartitionOverwriteCommand.scala#L32-L61) and the write-rule ordering in [PaimonAnalysis.scala](https://github.com/apache/paimon/blob/437bcc3c84a10b71b827a0a00c1661ccaad6e64f/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala#L45-L64). A dynamic <code>OverwritePartitionsDynamic</code> write without a <code>PARTITION (...)</code> clause does not exhibit this alias-driven misalignment at the baseline. That path only selects the Hive layout when the query output names already match the Hive-tail layout; otherwise it retains positional table order. #### Proposed fix direction The minimal fix is to improve the existing automatic decision before applying the Hive-tail rewrite: 1. Require the query and target table to have the same arity. 2. Find the dynamic partition columns and their positions in the target table. 3. If those positions in the query already carry the matching dynamic partition column names, treat the query as table-ordered and do not apply the Hive-tail rewrite. 4. Otherwise, retain the current Hive-tail behavior. For this reproduction, <code>query.output(0) = ds</code> and <code>query.output(1) = part</code>, which proves that the dynamic partition columns are already at their target-table positions even though the unrelated <code>detail_ratio</code> expression has a different name. This automatic check cannot resolve every positional ambiguity. For example, if the target partition columns are <code>ds</code> and <code>part</code> but the query fields at those positions are still named <code>dt</code> and <code>pt</code>, names alone cannot prove whether the input is table-ordered or Hive-ordered. If maintainers want an explicit escape hatch for such ambiguous inputs, an optional configuration could expose: ~~~text spark.paimon.sql.dynamic-partition-column-order=AUTO|TABLE|HIVE ~~~ - <code>AUTO</code>: use the improved automatic detection and remain the default. - <code>TABLE</code>: interpret <code>query.output(i)</code> as <code>table.output(i)</code>. - <code>HIVE</code>: require the dynamic partition columns at the end of the query. - <code>INSERT ... BY NAME</code>: do not apply this positional policy. The explicit configuration is not required for the minimal fix. #### Suggested test coverage 1. A UNION whose first branch has a non-partition alias mismatch while positions follow table order. 2. A single SELECT with the same table-order condition. 3. A genuine Hive-tail dynamic partition write remains supported. 4. A table whose dynamic partition columns are already trailing is not reordered again. 5. <code>AppendData</code>, <code>OverwriteByExpression</code>, and <code>OverwritePartitionsDynamic</code>, including dynamic overwrite with both values of <code>spark.paimon.write.use-v2-write</code>. 6. A no-<code>PARTITION (...)</code> <code>OverwritePartitionsDynamic</code> write and <code>INSERT ... BY NAME</code> remain unaffected. 7. If the optional mode is adopted: explicit <code>TABLE</code>, explicit <code>HIVE</code>, and invalid configuration values. ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- 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]
